Celery переход на quorum очередь

import os import re import html import logging from typing import Optional

import requests from dotenv import load_dotenv from telegram import Update from telegram.ext import ( ApplicationBuilder, CommandHandler, ContextTypes, )

load_dotenv()

BOT_TOKEN = os.getenv("BOT_TOKEN") ALLOWED_USER_ID = int(os.getenv("ALLOWED_USER_ID", "0"))

QBT_HOST = os.getenv("QBT_HOST", "").rstrip("/") QBT_USER = os.getenv("QBT_USER") QBT_PASS = os.getenv("QBT_PASS")

MOVIES_CATEGORY = os.getenv("MOVIES_CATEGORY", "movies") SERIES_CATEGORY = os.getenv("SERIES_CATEGORY", "series")

MAGNET_RE = re.compile(r"(magnet:\?xt=urn:[^\s]+)", re.IGNORECASE)

logging.basicConfig( format="%(asctime)s %(levelname)s %(name)s: %(message)s", level=logging.INFO, ) log = logging.getLogger("torrentbot")

def is_allowed(update: Update) -> bool: user = update.effective_user return bool(user and user.id == ALLOWED_USER_ID)

async def deny(update: Update): await update.message.reply_text("⛔ Не-а, факбойчик!.")

def qbt_session() -> requests.Session: session = requests.Session() r = session.post( f"{QBT_HOST}/api/v2/auth/login", data={ "username": QBT_USER, "password": QBT_PASS, }, timeout=15, )

if r.status_code != 200 or r.text.strip() != "Ok.":
    raise RuntimeError(f"qBittorrent login failed: {r.status_code} {r.text}")

return session

def extract_magnet(text: str) -> Optional[str]: match = MAGNET_RE.search(text or "") if not match: return None return match.group(1)

def qbt_add_magnet(magnet: str, category: str) -> None: session = qbt_session()

r = session.post(
    f"{QBT_HOST}/api/v2/torrents/add",
    data={
        "urls": magnet,
        "category": category,
        "paused": "false",
    },
    timeout=30,
)

if r.status_code != 200:
    raise RuntimeError(f"Нихуя не получилось доабвить торрент: {r.status_code} {r.text}")

if r.text.strip().lower() not in ("ok.", "ok"):
    raise RuntimeError(f"Какая-то хуйня в ответе: {r.text}")

def qbt_list(limit: int = 10): session = qbt_session() r = session.get( f"{QBT_HOST}/api/v2/torrents/info", timeout=20, ) r.raise_for_status()

torrents = r.json()
torrents.sort(key=lambda x: x.get("added_on", 0), reverse=True)
return torrents[:limit]

def qbt_pause_all(): session = qbt_session() r = session.post( f"{QBT_HOST}/api/v2/torrents/pause", data={"hashes": "all"}, timeout=20, ) r.raise_for_status()

def qbt_resume_all(): session = qbt_session() r = session.post( f"{QBT_HOST}/api/v2/torrents/resume", data={"hashes": "all"}, timeout=20, ) r.raise_for_status()

def qbt_delete_by_hash(torrent_hash: str, delete_files: bool = False): session = qbt_session() r = session.post( f"{QBT_HOST}/api/v2/torrents/delete", data={ "hashes": torrent_hash, "deleteFiles": "true" if delete_files else "false", }, timeout=20, ) r.raise_for_status()

def fmt_size(num: float) -> str: for unit in ["B", "KB", "MB", "GB", "TB"]: if abs(num) < 1024: return f"{num:.1f} {unit}" num /= 1024 return f"{num:.1f} PB"

def fmt_speed(num: float) -> str: return fmt_size(num) + "/s"

def fmt_torrent(t: dict) -> str: name = html.escape(t.get("name", "unknown")) progress = float(t.get("progress", 0)) * 100 state = html.escape(t.get("state", "unknown")) size = fmt_size(float(t.get("size", 0))) dlspeed = fmt_speed(float(t.get("dlspeed", 0))) category = html.escape(t.get("category", "")) h = html.escape(t.get("hash", ""))[:10]

return (
    f"🎬 <b>{name}</b>\n"
    f"├ {progress:.1f}% | {state}\n"
    f"├ {size} | ↓ {dlspeed}\n"
    f"├ category: <code>{category}</code>\n"
    f"└ hash: <code>{h}</code>"
)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

await update.message.reply_text(
    "Готов.\n\n"
    "Команды:\n"
    "/film <magnet> — добавить ебучий фильм\n"
    "/series <magnet> — добавить ебучий сериал\n"
    "/list — последние говно-торренты\n"
    "/status — активные как пидор загрузки\n"
    "/pause — поставить все на паузу блять\n"
    "/resume — продолжить нахуй все\n"
    "/delete <hash> — удалить ебаный торрент без файлов\n"
    "/deletefiles <hash> — удалить торрент и файлы все ебучие"
)

async def add_movie(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

text = update.message.text
magnet = extract_magnet(text)

if not magnet:
    return await update.message.reply_text(
        "Пришли так нахуй!!!!!:\n/film magnet:?xt=urn:btih:..."
    )

try:
    qbt_add_magnet(magnet, MOVIES_CATEGORY)
    await update.message.reply_text(
        f"✅ Гей-порно добавлено в qBittorrent\nКатегория: {MOVIES_CATEGORY}"
    )
except Exception as e:
    log.exception("film add failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!: {e}")

async def add_series(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

text = update.message.text
magnet = extract_magnet(text)

if not magnet:
    return await update.message.reply_text(
        "Пришли так:\n/series magnet:?xt=urn:btih:..."
    )

try:
    qbt_add_magnet(magnet, SERIES_CATEGORY)
    await update.message.reply_text(
        f"✅ Гей-Сериал добавлен в qBittorrent\nКатегория: {SERIES_CATEGORY}"
    )
except Exception as e:
    log.exception("series add failed")
    await update.message.reply_text(f"❌ Ошибка: {e}")

async def list_torrents(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

try:
    torrents = qbt_list(limit=10)
    if not torrents:
        return await update.message.reply_text("Список пуст.")

    text = "\n\n".join(fmt_torrent(t) for t in torrents)
    await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
    log.exception("list failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!!: {e}")

async def status(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

try:
    torrents = qbt_list(limit=50)
    active = [
        t for t in torrents
        if t.get("state") not in ("pausedDL", "pausedUP", "stalledUP", "uploading")
    ]

    if not active:
        return await update.message.reply_text("Активных загрузок нет.")

    text = "\n\n".join(fmt_torrent(t) for t in active[:10])
    await update.message.reply_text(text, parse_mode="HTML")
except Exception as e:
    log.exception("status failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!: {e}")

async def pause_all(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

try:
    qbt_pause_all()
    await update.message.reply_text("⏸ Все торренты поставлены на паузу.")
except Exception as e:
    log.exception("pause failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!: {e}")

async def resume_all(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

try:
    qbt_resume_all()
    await update.message.reply_text("▶️ Все торренты продолжены.")
except Exception as e:
    log.exception("resume failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!!: {e}")

async def delete(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

if not context.args:
    return await update.message.reply_text(
        "Пришли hash:\n/delete abc123\n\nHash можно посмотреть в /list."
    )

torrent_hash = context.args[0]

try:
    qbt_delete_by_hash(torrent_hash, delete_files=False)
    await update.message.reply_text("🗑 Торрент удалён, файлы оставлены.")
except Exception as e:
    log.exception("delete failed")
    await update.message.reply_text(f"❌ Ошибка епта!!!1: {e}")

async def delete_files(update: Update, context: ContextTypes.DEFAULT_TYPE): if not is_allowed(update): return await deny(update)

if not context.args:
    return await update.message.reply_text(
        "Пришли hash:\n/deletefiles abc123\n\nОсторожно: удалит торрент и файлы."
    )

torrent_hash = context.args[0]

try:
    qbt_delete_by_hash(torrent_hash, delete_files=True)
    await update.message.reply_text("🧨 Торрент и файлы удалены.")
except Exception as e:
    log.exception("delete files failed")
    await update.message.reply_text(f"❌ Ошибка: {e}")

def main(): if not BOT_TOKEN: raise RuntimeError("BOT_TOKEN is empty") if not QBT_HOST: raise RuntimeError("QBT_HOST is empty")

app = ApplicationBuilder().token(BOT_TOKEN).build()

app.add_handler(CommandHandler("start", start))
app.add_handler(CommandHandler("film", add_movie))
app.add_handler(CommandHandler("movie", add_movie))
app.add_handler(CommandHandler("series", add_series))
app.add_handler(CommandHandler("serial", add_series))
app.add_handler(CommandHandler("list", list_torrents))
app.add_handler(CommandHandler("status", status))
app.add_handler(CommandHandler("pause", pause_all))
app.add_handler(CommandHandler("resume", resume_all))
app.add_handler(CommandHandler("delete", delete))
app.add_handler(CommandHandler("deletefiles", delete_files))

log.info("Bot started")
app.run_polling()

if name == "main": main()