Мой код camgirlfinder.net API через телеграмм бота

anrlfdl · 24 ноя 2024

Привет, увлекаюсь коддингом на питоне и решил слить в паблик один из своих инструментов, суть заключается в автоматизации запросов к https://camgirlfinder.net/ через своего бота в телеграмм, это поможет пересылать фото из диалогов или других каналов\ботов. Так же проверяет подписку на определенный канал в тг, для этого необходимо добавить его как администратора.Пишите если что то будет не ясно, рад помочь каждому в свое свободное время.
import os
import requests
import logging
import hashlib
from telegram import Update, ReplyKeyboardMarkup, KeyboardButton, ChatMember
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes
# Настройка логирования
logging.basicConfig(level=[URL='https://lolz.live/https:/logging.INFO']logging.INFO[/URL] )
logger = logging.getLogger(__name__)
# Токен Telegram-бота
BOT_TOKEN = "вот сюда копируй"
# ID канала, на который нужно проверить подписку
CHANNEL_ID = "@Вот тут пиши не убирая кавычек" # Замените на ваш канал
# Хранилище для хэшей загруженных фотографий
uploaded_photos_hashes = set()
# Функция для загрузки фото на сайт и получения результатов
def upload_photo(photo_path):
url = "[URL='https://api.camgirlfinder.net/search%22']https://api.camgirlfinder.net/search"[/URL] # URL для загрузки
with open(photo_path, 'rb') as photo_file:
files = {'file': photo_file}
response = [URL='https://requests.post']requests.post[/URL] (url, files=files)
return response.json() if response.status_code == 200 else None
# Функция для создания хэша изображения
def hash_image(image_path):
hasher = hashlib.md5()
with open(image_path, 'rb') as image_file:
hasher.update(image_[URL='https://file.read']file.read[/URL] ())
return hasher.hexdigest()
# Функция для создания клавиатуры
def create_keyboard():
keyboard = [
[KeyboardButton("Поиск фото")],
[KeyboardButton("Информация о боте")]
]
return ReplyKeyboardMarkup(keyboard, resize_keyboard=True)
# Функция для проверки подписки пользователя
async def is_user_subscribed(user_id, context):
chat_member = await [URL='https://context.bot']context.bot[/URL] .get_chat_member(CHANNEL_ID, user_id)
return chat_member.status in [ChatMember.MEMBER, ChatMember.ADMINISTRATOR]
# Обработчик команды /start
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.message.from_[URL='https://user.id']user.id[/URL]
if await is_user_subscribed(user_id, context):
await update.message.reply_text(
"Привет! Я бот для поиска информации по фото. Выберите действие:",
reply_markup=create_keyboard()
)
else:
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text=f"Пожалуйста, подпишитесь на наш канал {CHANNEL_ID} для использования бота.",
reply_markup=create_keyboard()
)
# Функция для отправки сообщения с просьбой подписаться
async def request_subscription(update: Update, context: ContextTypes.DEFAULT_TYPE):
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text=f"Пожалуйста, подпишитесь на наш канал {CHANNEL_ID} для использования бота."
)
# Обработчик текстовых сообщений
async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.message.from_[URL='https://user.id']user.id[/URL]
if not await is_user_subscribed(user_id, context):
await request_subscription(update, context) # Отправляем просьбу подписаться
return # Игнорируем сообщения, если пользователь не подписан
text = update.message.text
if text == "Поиск фото":
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text="Пожалуйста, отправьте мне фото для поиска.",
reply_markup=create_keyboard()
)
elif text == "Информация о боте":
info_text = (
"Этот бот позволяет вам загружать фото и получать информацию о них.\n"
"Просто отправьте фото, и бот попробует найти информацию на сайте."
)
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text=info_text,
reply_markup=create_keyboard()
)
# Обработчик фото
async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):
user_id = update.message.from_[URL='https://user.id']user.id[/URL]
if not await is_user_subscribed(user_id, context):
await request_subscription(update, context) # Отправляем просьбу подписаться
return # Игнорируем фото, если пользователь не подписан
photo_path = None
try:
photo_file = await [URL='https://update.message.photo']update.message.photo[/URL] [-1].get_file()
photo_path = f"photo_{photo_file.file_id}.jpg"
await photo_file.download_to_drive(photo_path)
photo_hash = hash_image(photo_path)
if photo_hash in uploaded_photos_hashes:
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text="Это фото уже было отправлено ранее. Пожалуйста, отправьте другое фото.",
reply_markup=create_keyboard()
)
return
uploaded_photos_hashes.add(photo_hash)
result = upload_photo(photo_path)
if result is not None:
await process_result(result, update, context)
else:
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text="Извините, не удалось найти информацию.",
reply_markup=create_keyboard()
)
except Exception as e:
logger.error(f"Ошибка при обработке фото: {e}")
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text="Произошла ошибка при обработке вашего фото.",
reply_markup=create_keyboard()
)
finally:
if photo_path and os.path.exists(photo_path):
os.remove(photo_path)
[URL='https://logger.info']logger.info[/URL] (f"Удален временный файл: {photo_path}")
async def process_result(result, update, context):
predictions = result.get('predictions', [])
if predictions:
for idx, prediction in enumerate(predictions):
model = prediction.get('model', 'Неизвестно')
gender = prediction.get('gender', 'Неизвестно')
probability = prediction.get('probability', 'Неизвестно')
profile_url = prediction['urls'].get('profile', 'Нет ссылки на профиль')
message = (
f"Совпадение № {idx + 1}:\n"
f"Ник модели: {model}\n"
f"Гендер: {gender}\n"
f"Вероятность: {probability}\n"
f"Ссылка на профиль: {profile_url}\n"
)
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text=message
)
else:
await [URL='https://context.bot']context.bot[/URL] .send_message(
chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,
text="Не найдено совпадений.",
reply_markup=create_keyboard()
)
# Основная функция для запуска бота
def main():
application = ApplicationBuilder().token(BOT_TOKEN).build()
application.add_handler(CommandHandler("start", start))
application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text_message))
application.add_handler(MessageHandler([URL='https://filters.PHOTO']filters.PHOTO[/URL] , handle_photo))
application.run_polling()
if __name__ == "__main__":
main()

Python
import os

import requests

import logging

import hashlib

from telegram import Update, ReplyKeyboardMarkup, KeyboardButton, ChatMember

from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes

# Настройка логирования

logging.basicConfig(level=[URL='https://lolz.live/https:/logging.INFO']logging.INFO[/URL] )

logger = logging.getLogger(__name__)

# Токен Telegram-бота

BOT_TOKEN = "вот сюда копируй"

# ID канала, на который нужно проверить подписку

CHANNEL_ID = "@Вот тут пиши не убирая кавычек" # Замените на ваш канал

# Хранилище для хэшей загруженных фотографий

uploaded_photos_hashes = set()

# Функция для загрузки фото на сайт и получения результатов

def upload_photo(photo_path):

url = "[URL='https://api.camgirlfinder.net/search%22']https://api.camgirlfinder.net/search"[/URL]  # URL для загрузки

    with open(photo_path, 'rb') as photo_file:

        files = {'file': photo_file}

response = [URL='https://requests.post']requests.post[/URL] (url, files=files)

 return response.json() if response.status_code == 200 else None

# Функция для создания хэша изображения

def hash_image(image_path):

    hasher = hashlib.md5()

    with open(image_path, 'rb') as image_file:

hasher.update(image_[URL='https://file.read']file.read[/URL] ())

    return hasher.hexdigest()

# Функция для создания клавиатуры

def create_keyboard():

    keyboard = [

        [KeyboardButton("Поиск фото")],

        [KeyboardButton("Информация о боте")]

    ]

 return ReplyKeyboardMarkup(keyboard, resize_keyboard=True)

# Функция для проверки подписки пользователя

async def is_user_subscribed(user_id, context):

chat_member = await [URL='https://context.bot']context.bot[/URL] .get_chat_member(CHANNEL_ID, user_id)

    return chat_member.status in [ChatMember.MEMBER, ChatMember.ADMINISTRATOR]

# Обработчик команды /start

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):

user_id = update.message.from_[URL='https://user.id']user.id[/URL] 

    if await is_user_subscribed(user_id, context):

        await update.message.reply_text(

            "Привет! Я бот для поиска информации по фото. Выберите действие:",

            reply_markup=create_keyboard()

        )

    else:

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

            text=f"Пожалуйста, подпишитесь на наш канал {CHANNEL_ID} для использования бота.",

            reply_markup=create_keyboard()

 )

# Функция для отправки сообщения с просьбой подписаться

async def request_subscription(update: Update, context: ContextTypes.DEFAULT_TYPE):

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

        text=f"Пожалуйста, подпишитесь на наш канал {CHANNEL_ID} для использования бота."

 )

# Обработчик текстовых сообщений

async def handle_text_message(update: Update, context: ContextTypes.DEFAULT_TYPE):

user_id = update.message.from_[URL='https://user.id']user.id[/URL] 

    if not await is_user_subscribed(user_id, context):

await request_subscription(update, context) # Отправляем просьбу подписаться

return # Игнорируем сообщения, если пользователь не подписан

 text = update.message.text

    if text == "Поиск фото":

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

            text="Пожалуйста, отправьте мне фото для поиска.",

            reply_markup=create_keyboard()

        )

    elif text == "Информация о боте":

        info_text = (

            "Этот бот позволяет вам загружать фото и получать информацию о них.\n"

            "Просто отправьте фото, и бот попробует найти информацию на сайте."

        )

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

            text=info_text,

            reply_markup=create_keyboard()

 )

# Обработчик фото

async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE):

user_id = update.message.from_[URL='https://user.id']user.id[/URL] 

    if not await is_user_subscribed(user_id, context):

await request_subscription(update, context)  # Отправляем просьбу подписаться

return # Игнорируем фото, если пользователь не подписан

photo_path = None

    try:

photo_file = await [URL='https://update.message.photo']update.message.photo[/URL] [-1].get_file()

        photo_path = f"photo_{photo_file.file_id}.jpg"

        await photo_file.download_to_drive(photo_path)

        photo_hash = hash_image(photo_path)

        if photo_hash in uploaded_photos_hashes:

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

                text="Это фото уже было отправлено ранее. Пожалуйста, отправьте другое фото.",

                reply_markup=create_keyboard()

            )

 return

uploaded_photos_hashes.add(photo_hash)

        result = upload_photo(photo_path)

        if result is not None:

            await process_result(result, update, context)

        else:

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

                text="Извините, не удалось найти информацию.",

                reply_markup=create_keyboard()

            )

    except Exception as e:

        logger.error(f"Ошибка при обработке фото: {e}")

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

            text="Произошла ошибка при обработке вашего фото.",

            reply_markup=create_keyboard()

        )

    finally:

        if photo_path and os.path.exists(photo_path):

            os.remove(photo_path)

[URL='https://logger.info']logger.info[/URL] (f"Удален временный файл: {photo_path}")

async def process_result(result, update, context):

    predictions = result.get('predictions', [])

    if predictions:

        for idx, prediction in enumerate(predictions):

            model = prediction.get('model', 'Неизвестно')

            gender = prediction.get('gender', 'Неизвестно')

            probability = prediction.get('probability', 'Неизвестно')

            profile_url = prediction['urls'].get('profile', 'Нет ссылки на профиль')

            message = (

                f"Совпадение № {idx + 1}:\n"

                f"Ник модели: {model}\n"

                f"Гендер: {gender}\n"

                f"Вероятность: {probability}\n"

                f"Ссылка на профиль: {profile_url}\n"

            )

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

                text=message

            )

    else:

await [URL='https://context.bot']context.bot[/URL] .send_message(

chat_id=[URL='https://update.message.chat.id']update.message.chat.id[/URL] ,

            text="Не найдено совпадений.",

            reply_markup=create_keyboard()

)

# Основная функция для запуска бота

def main():

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

    application.add_handler(CommandHandler("start", start))

    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_text_message))

application.add_handler(MessageHandler([URL='https://filters.PHOTO']filters.PHOTO[/URL] , handle_photo))

    application.run_polling()

if __name__ == "__main__":

 main()
данный скрипт имеет следующие зависимости:
pip install telegram
pip install telegram.ext

их следует ввести в командную строку Windows перед запуском скрипта(win+r, в открывшееся окно ввести cmd), а так же перезапустить вашу IDE чтобы все успешно подгрузилось.

W223 · 24 ноя 2024

anrlfdl коддингом
Нажмите, чтобы раскрыть...

это конечно сильно

Показать тулбар -> Вставить (+) -> Код -> ctrl+v

рандомли · 24 ноя 2024

ни форматирования, ни гит репозитория, ничего человеческого...

Посоветуйте что можно написать из ботов в тг

Бот для обратной связи с админкой | Обход спамблока.

Бот-помощник для скупки рекламы | Сверка статусов/закрепов + аватарок

DICEBOT для форума | AIOGRAM бот с красивым меню

Триггербот Valorant OPENSOURCE

[Актуально] Юзер Бот - авто покупка подарков телеграм

Софт для генерации OctoBrowser и активации промиков

Софт для смены паролей firstmail

Много мужских имён в .txt файле

Парсер игр с 100% скидками стим

Неподтвержденные заказы - напоминалка для FPC

Автостатус в ВК и автобио в Telegram c текущей песней Spotify

Скрипт для скачивания всей музыки ВК + с обложкой + с текстом MusixMatch или Genius

Простейший вечный онлайн ВКонтакте

Pyanty - Веб-автоматизация, модуль для управления долфином через Selenium, и Dolphin API

Emunium - модуль для имитации поведения человека в Selenium/Pyppeteer/Playwright и т.д.

Gram_tools | Утильки для упрощенной разработки ботов на aiogram 3

CryptoScan - python библиотека для мониторинга (приёма) крипто платежей

Телеграм бот для проведения розыгрышей

Скрипт бота сносера на сессиях

[TG DRAIN] Telegram дрейнер

FREE| Telegram Parser | Телеграмм парсер username

[ЛУЧШИЙ] Бот Автопродаж | Телеграм Магазин

Шаблон бота Telegram

Рандомный статус ВКонтакте

Telegram бот: Gemini пересказывает все голосовые сообщения, которые вам приходят в ЛС

Звонки через Userbot в телеграм

Какие вообще платежки можно подключить если <18?

Как можно войти в телеграм аккаунт с помощью Auth Key (HEX) и DC ID. Ищу софт

Автостатус ВКонтакте с вашими аудиозаписями

Мой код camgirlfinder.net API через телеграмм бота