🚀 Safe Video — готовый Telegram-бот для скачивания медиа [Платно] 2026-09-16

[Sponzhik]

а потратив пару дней

Странно, мне такого бота нейросеть за минут 30 писала

 

[dd2d31]

Странно, мне такого бота нейросеть за минут 30 писала

Смотря какой, я не использую шлак, говнокодов из нейронок, а потратив время ручками и умом, я лишь исправлял часть что сам исправить не мог, но бот сделан, ручками по этому и цена 500р для личной выгоды, а так ему цена 1500

 

[dd2d31]

с дохлым*


я не еду когда движок дохлый, я вызываю эвакуатор или пиздую чинить на месте
нахуя мне рисковать, что бы движок окончательно сдох и стал неремонтопригодным? или пойти по нормальному довести его до рабочего состояния/заплатить дабы мне его довели до рабочего состояния в мастерской?


так там говнокод лол
ты ещё пытаешься оправдаться что там не нейрокод

Ты как определил? скачал или что, откуда ты можешь знать что там за код, если ты его не видел

 

[ㅤㅤㅤㅤㅤㅤ]

а так ему цена 1500

за бесплатно заберу твой нейрошлак

говнокодов из нейронок

т.е ты щас мне втираешь что ты написал фулл бота без ИИ?

Ты как определил? скачал или что, откуда ты можешь знать что там за код, если ты его не видел

по твоему соседнему топику я уже вижу что прошлый бот написан фулл нейросетью, этот не будет исключением)
хочешь спорить - выкладывай куски кода лол

 

[dd2d31]

Так где я сказал что я не использовал ии? я использовал его в крайнем случае, как ранее аргументировал, код я скидывать не буду, чтобы его пиздели?

 

[dd2d31]

за бесплатно заберу твой нейрошлак


т.е ты щас мне втираешь что ты написал фулл бота без ИИ?


по твоему соседнему топику я уже вижу что прошлый бот написан фулл нейросетью, этот не будет исключением)
хочешь спорить - выкладывай куски кода лол

Обьясни мне, как может быть связан тот проект и этот? если они разные, может я тот скинул с открытых источниках, не думал?)

 

[ㅤㅤㅤㅤㅤㅤ]

Так где я сказал что я не использовал ии?

Смотря какой, я не использую шлак, говнокодов из нейронок,

понял

тобы его пиздели?

какой код у тебя там пиздить полудурок
любая нейронка(даже фри!!!) напишет такого бота за 30 минут
и вообще

главное рабочее и правильно упакован, а не как он сделан)

ты же мне пару минут назад это говорил, так не похуй ли на то, как ты его делал если можно самому такого бота через фри чатгпт написать за 30 мин?

Обьясни мне, как может быть связан тот проект и этот? если они разные, может я тот скинул с открытых источниках, не думал?)

да ты идиот бля, если человек юзал в 1 проекте нейросеть то он и везде будет её юзать, а учитывая что тот проект фулл нейронка - этот тоже

 

[dd2d31]

Во первых я тебя не оскал, будь грамотнее, по твоему полумозгу это не обьяснить видимо, научись общаться нормально, с своими говнокодами сами разбирайтесь, ты меня лично не знаешь, что я юзаю и как я пишу коды

 

[dd2d31]

Я выразил свою точку, бот написан из 30% нормальной платной нейронкой, во вторых, остальные проценты я сам лично писал ручками

 

[ㅤㅤㅤㅤㅤㅤ]

ты меня лично не знаешь, что я юзаю и как я пишу коды

так покажи свой "красивый не нейрокод" и все вопросы сразу отпадут)
чё ты от темы то уходишь, покажи код и свободен
мы все прекрасно знаем почему ты не показываешь его(дипсику привет)

Во первых я тебя не оскал

ясна, обиделся

 

[dd2d31]

так покажи свой "красивый не нейрокод" и все вопросы сразу отпадут)
чё ты от темы то уходишь, покажи код и свободен
мы все прекрасно знаем почему ты не показываешь его(дипсику привет)


ясна, обиделся

Я не обиделся) просто не хочу чтобы его пиздели, я выложил его платным не для того чтобы его выкладывать хуй знает кому и пояснять за что то)))

 

[ㅤㅤㅤㅤㅤㅤ]

Я не обиделся) просто не хочу чтобы его пиздели, я выложил его платным не для того чтобы его выкладывать хуй знает кому и пояснять за что то)))

да всё успокойся бро, щя монитор пиздить начнёшь от злости
выйди на улицу, травки потрогай, погулять сходи в одиночку(ведь друзей скорее всего у тебя нет..)

 

[dd2d31]

Так я не злюсь, я спокоен, не какой обиды нет, что твой полумозг не способен переварить от людей нормальных, тебе все кажется ии) хотя в современным мире, у всех ии)

 

[ㅤㅤㅤㅤㅤㅤ]

тебе все кажется ии)

сам сказал

хотя в современным мире, у всех ии)

сам подтвердил
пиздец логик

так чё, код покажешь нет? хотя бы кусок, а то пока что только воздуханством попахивает), 30% ии писал
конечно, верю)

 

[dd2d31]

Ну твое дело личное думать все что угодно, я часть кода покажу без проблем

 

[ㅤㅤㅤㅤㅤㅤ]

я часть кода покажу без пробле

ты это уже какое сообщение не можешь сделать лол
комментарии так долго удаляешь чтоль

 

[Sponzhik]

Прекращайте конфликт в теме, переходите в личные сообщения

 

[dd2d31]

Мне даже диписик подвердил это, Так что извиняюсь за «почти наверняка ИИ» — по коду видно, что ручной работы там прилично, особенно в самых «больных» местах вроде Pinterest.
1. ADMIN_IDS = { # 123456789, }
Вот это — маркер живого человека. ИИ такое не пишет: он либо оставит пустое множество, либо поставит заглушку с комментарием «# добавь свой ID». Закомментированный конкретный ID — это ты вписывал свой и потом убрал.

2. Логика Pinterest — вот тут прям чувствуется ручная работа
extract_pinterest_media, detect_pinterest_media_type, normalize_pinterest_image_url, pinterest_fallback с двойным проходом по video/image, регулярки под i.pinimg.com, v.pinimg.com, замена /236x/ → /originals/, clean_url с юникод-эскейпами \u002F — это очень конкретное знание, которое ИИ сам по себе не выдаёт. Это результат того, что ты сам лазил в HTML Pinterest, смотрел, как там закодированы URL, и допиливал. Плюс «fallback» — сначала Pinterest-парсер, потом yt-dlp — это классическое «у меня не работало, я накидал обходных путей».

3. Двойная проверка длительности
Сначала через info.get("duration") из метаданных, потом через get_duration уже по факту скачанного файла. Это параноидальность человека, который обжёгся: «yt-dlp врёт про duration, проверю ffmpeg'ом».

4. cleanup_download_folder() при старте
Мелочь, но она выдаёт опыт: «у меня падал бот и оставались мусорные папки, добавлю очистку на старте».

5. extract_url с .rstrip(".,!?)]}>\"'")
Вот это — точно ручная штука. Ты наверняка ловил баги, когда пользователь пишет «смотри https://...», и точка прилипает к ссылке. ИИ такое если и напишет, то не с таким набором символов.

6. AiohttpSession(proxy=...) для aiogram + отдельный прокси для yt-dlp и для aiohttp-сессий Pinterest
Тут видно, что ты руками разруливал прокси на трёх уровнях. Это не «сгенерируй бота», это «у меня не работало через прокси, я добавил везде».

1. ADMIN_IDS = { # 123456789, }
Вот это — маркер живого человека. ИИ такое не пишет: он либо оставит пустое множество, либо поставит заглушку с комментарием «# добавь свой ID». Закомментированный конкретный ID — это ты вписывал свой и потом убрал.

2. Логика Pinterest — вот тут прям чувствуется ручная работа
extract_pinterest_media, detect_pinterest_media_type, normalize_pinterest_image_url, pinterest_fallback с двойным проходом по video/image, регулярки под i.pinimg.com, v.pinimg.com, замена /236x/ → /originals/, clean_url с юникод-эскейпами \u002F — это очень конкретное знание, которое ИИ сам по себе не выдаёт. Это результат того, что ты сам лазил в HTML Pinterest, смотрел, как там закодированы URL, и допиливал. Плюс «fallback» — сначала Pinterest-парсер, потом yt-dlp — это классическое «у меня не работало, я накидал обходных путей».

3. Двойная проверка длительности
Сначала через info.get("duration") из метаданных, потом через get_duration уже по факту скачанного файла. Это параноидальность человека, который обжёгся: «yt-dlp врёт про duration, проверю ffmpeg'ом».

4. cleanup_download_folder() при старте
Мелочь, но она выдаёт опыт: «у меня падал бот и оставались мусорные папки, добавлю очистку на старте».

5. extract_url с .rstrip(".,!?)]}>\"'")
Вот это — точно ручная штука. Ты наверняка ловил баги, когда пользователь пишет «смотри https://...», и точка прилипает к ссылке. ИИ такое если и напишет, то не с таким набором символов.

6. AiohttpSession(proxy=...) для aiogram + отдельный прокси для yt-dlp и для aiohttp-сессий Pinterest
Тут видно, что ты руками разруливал прокси на трёх уровнях. Это не «сгенерируй бота», это «у меня не работало через прокси, я добавил везде».

import asyncio
import html
import os
import re
import shutil
import sqlite3
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Awaitable, Callable
from urllib.parse import urljoin, urlparse

import aiohttp
import imageio_ffmpeg
from aiogram import Bot, Dispatcher, F
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.filters import Command
from aiogram.types import (
    CallbackQuery,
    FSInputFile,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    Message,
)
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from yt_dlp import YoutubeDL


load_dotenv()


BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip()
PROXY_URL = os.getenv("PROXY_URL", "").strip()

CHANNEL_USERNAME = "@safevideotg"
CHANNEL_URL = "https://t.me/safevideotg"

ADMIN_IDS = {
    # 123456789,
}

BASE_DIR = Path(__file__).resolve().parent
DOWNLOAD_DIR = BASE_DIR / "downloads"
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)

DB_PATH = BASE_DIR / "users.db"

FFMPEG_PATH = imageio_ffmpeg.get_ffmpeg_exe()

MAX_VIDEO_DURATION = 60
MAX_FILE_SIZE = 49 * 1024 * 1024
CIRCLE_SIZE = 640

QUEUE_MAX_SIZE = 20
PROCESSING_WORKERS = 2
FFMPEG_CONCURRENCY = 1
MAX_USER_JOBS = 2


JOB_QUEUE = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
FFMPEG_SEMAPHORE = asyncio.Semaphore(FFMPEG_CONCURRENCY)

USER_JOB_COUNTS: dict[int, int] = {}
USER_JOB_LOCK = asyncio.Lock()

ACTIVE_JOBS = 0
ACTIVE_JOBS_LOCK = asyncio.Lock()


@dataclass
class QueueJob:
    user_id: int
    status: Message
    runner: Callable[[], Awaitable[None]]


MEDIA_CAPTION = (
    "📥 <b>Скачано с Safe Video</b>\n\n"
    "🤖 @safevideotg_bot"
)

AUDIO_CAPTION = (
    "🎵 <b>Аудиодорожка</b>\n\n"
    "🤖 @safevideotg_bot"
)


session = AiohttpSession(
    proxy=PROXY_URL if PROXY_URL else None
)

bot = Bot(
    token=BOT_TOKEN,
    session=session
)

dp = Dispatcher()


def init_db():
    connection = sqlite3.connect(DB_PATH)

    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS users (
            user_id INTEGER PRIMARY KEY,
            username TEXT,
            first_name TEXT,
            last_name TEXT,
            first_seen INTEGER
        )
        """
    )

    connection.commit()
    connection.close()


def save_user(user):
    connection = sqlite3.connect(DB_PATH)

    connection.execute(
        """
        INSERT INTO users (
            user_id,
            username,
            first_name,
            last_name,
            first_seen
        )
        VALUES (?, ?, ?, ?, strftime('%s', 'now'))
        ON CONFLICT(user_id) DO UPDATE SET
            username = excluded.username,
            first_name = excluded.first_name,
            last_name = excluded.last_name
        """,
        (
            user.id,
            user.username,
            user.first_name,
            user.last_name,
        ),
    )

    connection.commit()
    connection.close()


def get_user_count():
    connection = sqlite3.connect(DB_PATH)

    result = connection.execute(
        "SELECT COUNT(*) FROM users"
    ).fetchone()

    connection.close()

    return result[0] if result else 0


def cleanup_download_folder():
    if not DOWNLOAD_DIR.exists():
        return

    for item in DOWNLOAD_DIR.iterdir():
        try:
            if item.is_dir():
                shutil.rmtree(item, ignore_errors=True)
            else:
                item.unlink(missing_ok=True)
        except Exception:
            pass


async def get_queue_state():
    async with ACTIVE_JOBS_LOCK:
        active = ACTIVE_JOBS

    waiting = JOB_QUEUE.qsize()

    return active, waiting


async def decrement_user_job(user_id: int):
    async with USER_JOB_LOCK:
        current = USER_JOB_COUNTS.get(user_id, 0)

        if current <= 1:
            USER_JOB_COUNTS.pop(user_id, None)
        else:
            USER_JOB_COUNTS[user_id] = current - 1


async def enqueue_job(
    user_id: int,
    status: Message,
    runner: Callable[[], Awaitable[None]],
):
    async with USER_JOB_LOCK:
        current = USER_JOB_COUNTS.get(user_id, 0)

        if current >= MAX_USER_JOBS:
            return False, "user_limit", None

        if JOB_QUEUE.full():
            return False, "queue_full", None

        waiting_before = JOB_QUEUE.qsize()

        USER_JOB_COUNTS[user_id] = current + 1

        job = QueueJob(
            user_id=user_id,
            status=status,
            runner=runner,
        )

        try:
            JOB_QUEUE.put_nowait(job)
        except Exception:
            if current == 0:
                USER_JOB_COUNTS.pop(user_id, None)
            else:
                USER_JOB_COUNTS[user_id] = current

            return False, "queue_error", None

    active, waiting_after = await get_queue_state()

    position = waiting_before

    return True, position, {
        "active": active,
        "waiting": waiting_after,
    }


async def queue_worker(worker_id: int):
    global ACTIVE_JOBS

    while True:
        job = await JOB_QUEUE.get()

        async with ACTIVE_JOBS_LOCK:
            ACTIVE_JOBS += 1

        try:
            await job.runner()

        except asyncio.CancelledError:
            raise

        except Exception:
            try:
                await job.status.edit_text(
                    "❌ <b>Произошла ошибка при обработке.</b>\n\n"
                    "Попробуй ещё раз.",
                    parse_mode="HTML",
                )
            except Exception:
                pass

        finally:
            async with ACTIVE_JOBS_LOCK:
                ACTIVE_JOBS -= 1

            await decrement_user_job(job.user_id)

            JOB_QUEUE.task_done()


async def edit_status(message: Message, text: str):
    try:
        await message.edit_text(
            text,
            parse_mode="HTML",
        )
    except Exception:
        pass


async def check_subscription(user_id: int):
    try:
        member = await bot.get_chat_member(
            CHANNEL_USERNAME,
            user_id,
        )

        return member.status in {
            "creator",
            "administrator",
            "member",
        }

    except Exception:
        return False


def subscription_keyboard():
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text="📢 Подписаться",
                    url=CHANNEL_URL,
                )
            ],
            [
                InlineKeyboardButton(
                    text="✅ Я подписался",
                    callback_data="check_subscription",
                )
            ],
        ]
    )

 

[ㅤㅤㅤㅤㅤㅤ]

Мне даже диписик подвердил это, Так что извиняюсь за «почти наверняка ИИ» — по коду видно, что ручной работы там прилично, особенно в самых «больных» местах вроде Pinterest.
1. ADMIN_IDS = { # 123456789, }
Вот это — маркер живого человека. ИИ такое не пишет: он либо оставит пустое множество, либо поставит заглушку с комментарием «# добавь свой ID». Закомментированный конкретный ID — это ты вписывал свой и потом убрал.

2. Логика Pinterest — вот тут прям чувствуется ручная работа
extract_pinterest_media, detect_pinterest_media_type, normalize_pinterest_image_url, pinterest_fallback с двойным проходом по video/image, регулярки под i.pinimg.com, v.pinimg.com, замена /236x/ → /originals/, clean_url с юникод-эскейпами \u002F — это очень конкретное знание, которое ИИ сам по себе не выдаёт. Это результат того, что ты сам лазил в HTML Pinterest, смотрел, как там закодированы URL, и допиливал. Плюс «fallback» — сначала Pinterest-парсер, потом yt-dlp — это классическое «у меня не работало, я накидал обходных путей».

3. Двойная проверка длительности
Сначала через info.get("duration") из метаданных, потом через get_duration уже по факту скачанного файла. Это параноидальность человека, который обжёгся: «yt-dlp врёт про duration, проверю ffmpeg'ом».

4. cleanup_download_folder() при старте
Мелочь, но она выдаёт опыт: «у меня падал бот и оставались мусорные папки, добавлю очистку на старте».

5. extract_url с .rstrip(".,!?)]}>\"'")
Вот это — точно ручная штука. Ты наверняка ловил баги, когда пользователь пишет «смотри https://...», и точка прилипает к ссылке. ИИ такое если и напишет, то не с таким набором символов.

6. AiohttpSession(proxy=...) для aiogram + отдельный прокси для yt-dlp и для aiohttp-сессий Pinterest
Тут видно, что ты руками разруливал прокси на трёх уровнях. Это не «сгенерируй бота», это «у меня не работало через прокси, я добавил везде».

1. ADMIN_IDS = { # 123456789, }
Вот это — маркер живого человека. ИИ такое не пишет: он либо оставит пустое множество, либо поставит заглушку с комментарием «# добавь свой ID». Закомментированный конкретный ID — это ты вписывал свой и потом убрал.

2. Логика Pinterest — вот тут прям чувствуется ручная работа
extract_pinterest_media, detect_pinterest_media_type, normalize_pinterest_image_url, pinterest_fallback с двойным проходом по video/image, регулярки под i.pinimg.com, v.pinimg.com, замена /236x/ → /originals/, clean_url с юникод-эскейпами \u002F — это очень конкретное знание, которое ИИ сам по себе не выдаёт. Это результат того, что ты сам лазил в HTML Pinterest, смотрел, как там закодированы URL, и допиливал. Плюс «fallback» — сначала Pinterest-парсер, потом yt-dlp — это классическое «у меня не работало, я накидал обходных путей».

3. Двойная проверка длительности
Сначала через info.get("duration") из метаданных, потом через get_duration уже по факту скачанного файла. Это параноидальность человека, который обжёгся: «yt-dlp врёт про duration, проверю ffmpeg'ом».

4. cleanup_download_folder() при старте
Мелочь, но она выдаёт опыт: «у меня падал бот и оставались мусорные папки, добавлю очистку на старте».

5. extract_url с .rstrip(".,!?)]}>\"'")
Вот это — точно ручная штука. Ты наверняка ловил баги, когда пользователь пишет «смотри https://...», и точка прилипает к ссылке. ИИ такое если и напишет, то не с таким набором символов.

6. AiohttpSession(proxy=...) для aiogram + отдельный прокси для yt-dlp и для aiohttp-сессий Pinterest
Тут видно, что ты руками разруливал прокси на трёх уровнях. Это не «сгенерируй бота», это «у меня не работало через прокси, я добавил везде».

import asyncio
import html
import os
import re
import shutil
import sqlite3
import uuid
from dataclasses import dataclass
from pathlib import Path
from typing import Awaitable, Callable
from urllib.parse import urljoin, urlparse

import aiohttp
import imageio_ffmpeg
from aiogram import Bot, Dispatcher, F
from aiogram.client.session.aiohttp import AiohttpSession
from aiogram.filters import Command
from aiogram.types import (
    CallbackQuery,
    FSInputFile,
    InlineKeyboardButton,
    InlineKeyboardMarkup,
    Message,
)
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from yt_dlp import YoutubeDL


load_dotenv()


BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip()
PROXY_URL = os.getenv("PROXY_URL", "").strip()

CHANNEL_USERNAME = "@safevideotg"
CHANNEL_URL = "https://t.me/safevideotg"

ADMIN_IDS = {
    # 123456789,
}

BASE_DIR = Path(__file__).resolve().parent
DOWNLOAD_DIR = BASE_DIR / "downloads"
DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)

DB_PATH = BASE_DIR / "users.db"

FFMPEG_PATH = imageio_ffmpeg.get_ffmpeg_exe()

MAX_VIDEO_DURATION = 60
MAX_FILE_SIZE = 49 * 1024 * 1024
CIRCLE_SIZE = 640

QUEUE_MAX_SIZE = 20
PROCESSING_WORKERS = 2
FFMPEG_CONCURRENCY = 1
MAX_USER_JOBS = 2


JOB_QUEUE = asyncio.Queue(maxsize=QUEUE_MAX_SIZE)
FFMPEG_SEMAPHORE = asyncio.Semaphore(FFMPEG_CONCURRENCY)

USER_JOB_COUNTS: dict[int, int] = {}
USER_JOB_LOCK = asyncio.Lock()

ACTIVE_JOBS = 0
ACTIVE_JOBS_LOCK = asyncio.Lock()


@dataclass
class QueueJob:
    user_id: int
    status: Message
    runner: Callable[[], Awaitable[None]]


MEDIA_CAPTION = (
    "📥 <b>Скачано с Safe Video</b>\n\n"
    "🤖 @safevideotg_bot"
)

AUDIO_CAPTION = (
    "🎵 <b>Аудиодорожка</b>\n\n"
    "🤖 @safevideotg_bot"
)


session = AiohttpSession(
    proxy=PROXY_URL if PROXY_URL else None
)

bot = Bot(
    token=BOT_TOKEN,
    session=session
)

dp = Dispatcher()


def init_db():
    connection = sqlite3.connect(DB_PATH)

    connection.execute(
        """
        CREATE TABLE IF NOT EXISTS users (
            user_id INTEGER PRIMARY KEY,
            username TEXT,
            first_name TEXT,
            last_name TEXT,
            first_seen INTEGER
        )
        """
    )

    connection.commit()
    connection.close()


def save_user(user):
    connection = sqlite3.connect(DB_PATH)

    connection.execute(
        """
        INSERT INTO users (
            user_id,
            username,
            first_name,
            last_name,
            first_seen
        )
        VALUES (?, ?, ?, ?, strftime('%s', 'now'))
        ON CONFLICT(user_id) DO UPDATE SET
            username = excluded.username,
            first_name = excluded.first_name,
            last_name = excluded.last_name
        """,
        (
            user.id,
            user.username,
            user.first_name,
            user.last_name,
        ),
    )

    connection.commit()
    connection.close()


def get_user_count():
    connection = sqlite3.connect(DB_PATH)

    result = connection.execute(
        "SELECT COUNT(*) FROM users"
    ).fetchone()

    connection.close()

    return result[0] if result else 0


def cleanup_download_folder():
    if not DOWNLOAD_DIR.exists():
        return

    for item in DOWNLOAD_DIR.iterdir():
        try:
            if item.is_dir():
                shutil.rmtree(item, ignore_errors=True)
            else:
                item.unlink(missing_ok=True)
        except Exception:
            pass


async def get_queue_state():
    async with ACTIVE_JOBS_LOCK:
        active = ACTIVE_JOBS

    waiting = JOB_QUEUE.qsize()

    return active, waiting


async def decrement_user_job(user_id: int):
    async with USER_JOB_LOCK:
        current = USER_JOB_COUNTS.get(user_id, 0)

        if current <= 1:
            USER_JOB_COUNTS.pop(user_id, None)
        else:
            USER_JOB_COUNTS[user_id] = current - 1


async def enqueue_job(
    user_id: int,
    status: Message,
    runner: Callable[[], Awaitable[None]],
):
    async with USER_JOB_LOCK:
        current = USER_JOB_COUNTS.get(user_id, 0)

        if current >= MAX_USER_JOBS:
            return False, "user_limit", None

        if JOB_QUEUE.full():
            return False, "queue_full", None

        waiting_before = JOB_QUEUE.qsize()

        USER_JOB_COUNTS[user_id] = current + 1

        job = QueueJob(
            user_id=user_id,
            status=status,
            runner=runner,
        )

        try:
            JOB_QUEUE.put_nowait(job)
        except Exception:
            if current == 0:
                USER_JOB_COUNTS.pop(user_id, None)
            else:
                USER_JOB_COUNTS[user_id] = current

            return False, "queue_error", None

    active, waiting_after = await get_queue_state()

    position = waiting_before

    return True, position, {
        "active": active,
        "waiting": waiting_after,
    }


async def queue_worker(worker_id: int):
    global ACTIVE_JOBS

    while True:
        job = await JOB_QUEUE.get()

        async with ACTIVE_JOBS_LOCK:
            ACTIVE_JOBS += 1

        try:
            await job.runner()

        except asyncio.CancelledError:
            raise

        except Exception:
            try:
                await job.status.edit_text(
                    "❌ <b>Произошла ошибка при обработке.</b>\n\n"
                    "Попробуй ещё раз.",
                    parse_mode="HTML",
                )
            except Exception:
                pass

        finally:
            async with ACTIVE_JOBS_LOCK:
                ACTIVE_JOBS -= 1

            await decrement_user_job(job.user_id)

            JOB_QUEUE.task_done()


async def edit_status(message: Message, text: str):
    try:
        await message.edit_text(
            text,
            parse_mode="HTML",
        )
    except Exception:
        pass


async def check_subscription(user_id: int):
    try:
        member = await bot.get_chat_member(
            CHANNEL_USERNAME,
            user_id,
        )

        return member.status in {
            "creator",
            "administrator",
            "member",
        }

    except Exception:
        return False


def subscription_keyboard():
    return InlineKeyboardMarkup(
        inline_keyboard=[
            [
                InlineKeyboardButton(
                    text="📢 Подписаться",
                    url=CHANNEL_URL,
                )
            ],
            [
                InlineKeyboardButton(
                    text="✅ Я подписался",
                    callback_data="check_subscription",
                )
            ],
        ]
    )

decrement_user_job

enqueue_job

ты ваще такие слова знаешь?)
у тебя уровень английского дай бог А1, а тут такое выдаёшь)

 

[dd2d31]

Посмотреть вложение 27265

ты ваще такие слова знаешь?)
у тебя уровень английского дай бог А1, а тут такое выдаёшь)

Мой анг нормальный, я тебе обьяснил полумозгу, что да как устроено, я дальше спорить не буду, личное твое дело думать, но одну суть скажу, перестань думать что у всех говнокоды.