Yuriy Yuriev
343 lines
13 KiB
Python
343 lines
13 KiB
Python
"""
|
||
Менеджер задач по стримерам.
|
||
Простой список с возможностью изменять параметры на лету.
|
||
"""
|
||
|
||
import asyncio
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional, List, Dict
|
||
from datetime import datetime
|
||
import random
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class TaskParams:
|
||
"""Динамические параметры задачи."""
|
||
url: str = ""
|
||
task_type: str = "visit"
|
||
|
||
# Задержки
|
||
min_delay: int = 10
|
||
max_delay: int = 30
|
||
min_reading: int = 5
|
||
max_reading: int = 15
|
||
min_click_delay: int = 5 # задержка между кликами внутри серии
|
||
max_click_delay: int = 15
|
||
|
||
# Лимиты
|
||
max_visits: Optional[int] = None
|
||
current_visit: int = 0
|
||
|
||
# Twitch параметры
|
||
channel: str = ""
|
||
target_username: str = ""
|
||
visits_per_link: int = 5
|
||
monitor_minutes: int = 10
|
||
allowed_domains: Optional[List[str]] = None
|
||
|
||
visits_percent: float = 0.0 # % от зрителей (0 = использовать visits_per_link)
|
||
min_series: int = 1 # для user_visit: мин. кликов в серии
|
||
max_series: int = 3 # для user_visit: макс. кликов в серии
|
||
min_ctr: float = 0.8 # для twitch_irc: мин. коэф. эффективности клика
|
||
max_ctr: float = 1.0 # для twitch_irc: макс. коэф. эффективности клика
|
||
|
||
# Состояние
|
||
paused: bool = False
|
||
stream_offline: bool = False # runtime: стрим оффлайн (не сохраняется)
|
||
auto_paused: bool = False # runtime: пауза выставлена автоматически (не сохраняется)
|
||
skip_next: bool = False
|
||
force_delay: Optional[int] = None
|
||
force_reading: Optional[int] = None
|
||
|
||
# Статистика
|
||
started_at: Optional[datetime] = None
|
||
total_visits: int = 0
|
||
successful_visits: int = 0
|
||
links_found: int = 0
|
||
chat_id: Optional[int] = None
|
||
user_id: Optional[int] = None # Telegram user_id, если задача создана пользователем
|
||
|
||
completed: bool = False # Задача завершена
|
||
stopped: bool = False # Задача остановлена пользователем
|
||
|
||
@property
|
||
def monitor_duration(self) -> int:
|
||
return self.monitor_minutes * 60
|
||
|
||
def update(self, **kwargs):
|
||
for key, value in kwargs.items():
|
||
if hasattr(self, key):
|
||
setattr(self, key, value)
|
||
|
||
def get_delay(self) -> int:
|
||
if self.skip_next:
|
||
self.skip_next = False
|
||
return 0
|
||
if self.force_delay:
|
||
delay = self.force_delay
|
||
self.force_delay = None
|
||
return delay
|
||
return random.randint(self.min_delay, self.max_delay)
|
||
|
||
def get_reading_time(self) -> int:
|
||
if self.force_reading:
|
||
reading = self.force_reading
|
||
self.force_reading = None
|
||
return reading
|
||
return random.randint(self.min_reading, self.max_reading)
|
||
|
||
def get_click_delay(self) -> int:
|
||
return random.randint(self.min_click_delay, self.max_click_delay)
|
||
|
||
def get_series_size(self) -> int:
|
||
return random.randint(self.min_series, self.max_series)
|
||
|
||
def get_status_emoji(self) -> str:
|
||
if self.stopped:
|
||
return "🛑"
|
||
if self.completed:
|
||
return "✅"
|
||
if self.paused:
|
||
return "⏸️"
|
||
return "🔄"
|
||
|
||
def get_status_text(self) -> str:
|
||
if self.stopped:
|
||
return "🛑 Остановлена"
|
||
if self.completed:
|
||
return "✅ Завершена"
|
||
if self.paused:
|
||
return "⏸️ Пауза"
|
||
return "🔄 Активна"
|
||
|
||
|
||
def get_stats_text(self) -> str:
|
||
"""Текст статистики."""
|
||
if self.task_type == "twitch_irc":
|
||
status = self.get_status_text()
|
||
duration = ""
|
||
if self.started_at and (self.completed or self.stopped):
|
||
delta = datetime.now() - self.started_at
|
||
duration = f"├ 🕐 Длилась: {delta.seconds // 60} мин\n"
|
||
|
||
domains_text = (
|
||
f"├ 🌐 Домены: {', '.join(self.allowed_domains)}\n"
|
||
if self.allowed_domains else
|
||
"├ 🌐 Домены: все (без фильтра)\n"
|
||
)
|
||
ctr_str = (
|
||
f"{self.min_ctr}%–{self.max_ctr}%"
|
||
if self.min_ctr != self.max_ctr
|
||
else f"{self.min_ctr}%"
|
||
)
|
||
ctr_line = f"├ 📊 CTR: {ctr_str} (зрители × CTR% = кликов)"
|
||
monitor = (
|
||
f"{self.monitor_minutes} мин" if self.monitor_minutes > 0 else "бесконечно"
|
||
)
|
||
|
||
return (
|
||
f"📺 **{self.channel}**\n"
|
||
f"├ Статус: {status}\n"
|
||
f"├ 🔗 Ссылок найдено в чате: {self.links_found}\n"
|
||
f"├ 📊 Всего кликов: {self.total_visits}\n"
|
||
f"├ ✅ Успешных кликов: {self.successful_visits}\n"
|
||
f"{ctr_line}\n"
|
||
f"├ 🔗 Резерв (без GQL): {self.visits_per_link}\n"
|
||
f"├ ⏱️ Пауза между переходами: {self.min_delay}–{self.max_delay} сек\n"
|
||
f"├ 📖 Время на странице: {self.min_reading}–{self.max_reading} сек\n"
|
||
f"├ ⏰ Длительность мониторинга: {monitor}\n"
|
||
f"{domains_text}"
|
||
f"{duration}"
|
||
)
|
||
else:
|
||
series_range = (
|
||
f"{self.min_series}" if self.min_series == self.max_series
|
||
else f"{self.min_series}–{self.max_series}"
|
||
)
|
||
return (
|
||
f"🌐 `{self.url[:50]}`\n"
|
||
f"├ Статус: {self.get_status_text()}\n"
|
||
f"├ 📊 Выполнено: {self.successful_visits}/{self.max_visits} переходов\n"
|
||
f"├ 🔢 Кликов в серии: {series_range}\n"
|
||
f"├ ⏳ Задержка между кликами: {self.min_click_delay}–{self.max_click_delay} сек\n"
|
||
f"├ ⏱️ Пауза между сериями: {self.min_delay}–{self.max_delay} сек\n"
|
||
f"├ 📖 Время на странице: {self.min_reading}–{self.max_reading} сек\n"
|
||
)
|
||
|
||
|
||
|
||
class TaskManager:
|
||
"""
|
||
Менеджер задач.
|
||
Хранит список стримеров и позволяет управлять каждым.
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._tasks: Dict[str, TaskParams] = {} # task_id -> params
|
||
self._lock = asyncio.Lock()
|
||
|
||
async def add_task(self, task_id: str, params: TaskParams) -> None:
|
||
"""Добавляет задачу."""
|
||
async with self._lock:
|
||
self._tasks[task_id] = params
|
||
logger.info(f"➕ Задача: {task_id[:12]} ({params.channel or params.url})")
|
||
|
||
async def remove_task(self, task_id: str) -> bool:
|
||
"""Удаляет задачу."""
|
||
async with self._lock:
|
||
if task_id in self._tasks:
|
||
del self._tasks[task_id]
|
||
logger.info(f"➖ Задача удалена: {task_id[:12]}")
|
||
return True
|
||
return False
|
||
|
||
async def get_task(self, task_id: str) -> Optional[TaskParams]:
|
||
"""Получает задачу."""
|
||
async with self._lock:
|
||
return self._tasks.get(task_id)
|
||
|
||
async def update_task(self, task_id: str, **kwargs) -> bool:
|
||
"""Обновляет параметры задачи."""
|
||
params = await self.get_task(task_id)
|
||
if params:
|
||
params.update(**kwargs)
|
||
return True
|
||
return False
|
||
|
||
async def complete_task(self, task_id: str) -> bool:
|
||
"""Пометить задачу как завершенную."""
|
||
return await self.update_task(task_id, completed=True, paused=True)
|
||
|
||
async def stop_task(self, task_id: str) -> bool:
|
||
"""Пометить задачу как остановленную."""
|
||
return await self.update_task(task_id, stopped=True, paused=True)
|
||
|
||
async def get_active_tasks(self) -> Dict[str, TaskParams]:
|
||
"""Только активные задачи (не завершенные и не остановленные)."""
|
||
return {
|
||
tid: p for tid, p in self._tasks.items()
|
||
if not p.completed and not p.stopped
|
||
}
|
||
|
||
async def get_completed_tasks(self) -> Dict[str, TaskParams]:
|
||
"""Завершенные задачи (для просмотра статистики)."""
|
||
return {
|
||
tid: p for tid, p in self._tasks.items()
|
||
if p.completed or p.stopped
|
||
}
|
||
|
||
|
||
async def pause_task(self, task_id: str) -> bool:
|
||
"""Пауза задачи."""
|
||
return await self.update_task(task_id, paused=True)
|
||
|
||
async def resume_task(self, task_id: str) -> bool:
|
||
"""Снять с паузы."""
|
||
return await self.update_task(task_id, paused=False)
|
||
|
||
async def skip_next_task(self, task_id: str) -> bool:
|
||
"""Пропустить следующее действие."""
|
||
return await self.update_task(task_id, skip_next=True)
|
||
|
||
async def restart_task(self, task_id: str) -> bool:
|
||
"""Сбросить счетчики для перезапуска."""
|
||
return await self.update_task(
|
||
task_id,
|
||
current_visit=0,
|
||
total_visits=0,
|
||
successful_visits=0,
|
||
links_found=0
|
||
)
|
||
|
||
async def get_all_tasks(self) -> Dict[str, TaskParams]:
|
||
"""Все задачи."""
|
||
async with self._lock:
|
||
return dict(self._tasks)
|
||
|
||
async def get_tasks_by_type(self, task_type: str) -> Dict[str, TaskParams]:
|
||
"""Задачи по типу."""
|
||
return {
|
||
tid: p for tid, p in self._tasks.items()
|
||
if p.task_type == task_type
|
||
}
|
||
|
||
async def get_twitch_tasks(self) -> Dict[str, TaskParams]:
|
||
"""Только Twitch задачи."""
|
||
return await self.get_tasks_by_type("twitch_irc")
|
||
|
||
async def get_visit_tasks(self) -> Dict[str, TaskParams]:
|
||
"""Только задачи посещения."""
|
||
return await self.get_tasks_by_type("visit")
|
||
|
||
async def get_stats(self) -> dict:
|
||
"""Общая статистика."""
|
||
twitch = visit = paused = active = 0
|
||
for p in self._tasks.values():
|
||
if p.task_type == "twitch_irc":
|
||
twitch += 1
|
||
elif p.task_type == "visit":
|
||
visit += 1
|
||
if p.paused:
|
||
paused += 1
|
||
else:
|
||
active += 1
|
||
return {
|
||
"total": len(self._tasks),
|
||
"twitch": twitch,
|
||
"visit": visit,
|
||
"paused": paused,
|
||
"active": active,
|
||
}
|
||
|
||
def _get_task_info(self, task_id: str, params: TaskParams) -> str:
|
||
"""Информация о задаче для отображения."""
|
||
status = params.get_status_emoji()
|
||
|
||
if params.task_type == "twitch_irc":
|
||
return (
|
||
f"{status} 📺 **{params.channel}**\n"
|
||
f" `{task_id[:12]}...`\n"
|
||
f" 👤 @{params.target_username} | "
|
||
f"🔗{params.links_found} | "
|
||
f"📊{params.total_visits}\n"
|
||
)
|
||
else:
|
||
return (
|
||
f"{status} 🌐 `{params.url[:40]}`\n"
|
||
f" `{task_id[:12]}...`\n"
|
||
f" 📊{params.total_visits} визитов\n"
|
||
)
|
||
|
||
async def format_task_list(self, max_items: int = 15) -> str:
|
||
"""Форматирует список задач для сообщения."""
|
||
async with self._lock:
|
||
if not self._tasks:
|
||
return "📊 Нет активных задач"
|
||
|
||
# Сначала Twitch, потом визиты
|
||
twitch = await self.get_twitch_tasks()
|
||
visits = await self.get_visit_tasks()
|
||
|
||
async with self._lock:
|
||
total_tasks = len(self._tasks)
|
||
text = f"📊 **ЗАДАЧИ: {total_tasks}**\n\n"
|
||
|
||
if twitch:
|
||
text += "📺 **Twitch мониторинг:**\n"
|
||
for i, (task_id, params) in enumerate(list(twitch.items())[:5]):
|
||
text += self._get_task_info(task_id, params)
|
||
if len(twitch) > 5:
|
||
text += f" _...и еще {len(twitch) - 5}_\n"
|
||
text += "\n"
|
||
|
||
if visits:
|
||
text += "🌐 **Посещение страниц:**\n"
|
||
for i, (task_id, params) in enumerate(list(visits.items())[:5]):
|
||
text += self._get_task_info(task_id, params)
|
||
if len(visits) > 5:
|
||
text += f" _...и еще {len(visits) - 5}_\n"
|
||
|
||
return text |