Files
Click/managers/task_manager.py
T
2026-05-18 18:35:09 +07:00

343 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
Менеджер задач по стримерам.
Простой список с возможностью изменять параметры на лету.
"""
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