Spaces:
Running
Running
import os | |
import logging | |
import requests | |
import base64 | |
from PIL import Image | |
from io import BytesIO | |
from telegram import Update, InputFile | |
from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackContext | |
# Настройка логирования | |
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") | |
logger = logging.getLogger(__name__) | |
# URL API | |
BASE_URL = "https://faceonlive-face-search-online.hf.space/api/predict" | |
def image_to_base64(image): | |
"""Конвертирует изображение в base64.""" | |
buffered = BytesIO() | |
image.save(buffered, format="JPEG") # JPEG для уменьшения размера | |
return base64.b64encode(buffered.getvalue()).decode('utf-8') | |
def compress_image(image, max_size=(512, 512)): | |
"""Уменьшает изображение до заданного размера.""" | |
image.thumbnail(max_size, Image.ANTIALIAS) | |
return image | |
def search_face(file_bytes): | |
"""Отправляет изображение в API и получает результат.""" | |
try: | |
response = requests.post( | |
BASE_URL, | |
files={"file": ("image.jpg", file_bytes)}, | |
headers={"accept": "application/json"} | |
) | |
response.raise_for_status() | |
return response.json() | |
except requests.exceptions.RequestException as e: | |
logger.error(f"Ошибка при запросе к API: {e}") | |
return {"error": str(e)} | |
def start(update: Update, context: CallbackContext): | |
"""Обрабатывает команду /start.""" | |
update.message.reply_text("Привет! Отправьте изображение, чтобы найти совпадения.") | |
def handle_image(update: Update, context: CallbackContext): | |
"""Обрабатывает изображения, отправленные пользователями.""" | |
photo = update.message.photo[-1].get_file() | |
file_bytes = BytesIO() | |
photo.download(out=file_bytes) | |
try: | |
# Открываем и сжимаем изображение | |
image = Image.open(file_bytes) | |
image = compress_image(image) | |
# Конвертируем изображение в байты | |
compressed_bytes = BytesIO() | |
image.save(compressed_bytes, format="JPEG") | |
compressed_bytes.seek(0) | |
# Отправляем изображение в API | |
result = search_face(compressed_bytes.getvalue()) | |
except Exception as e: | |
update.message.reply_text(f"Ошибка обработки изображения: {e}") | |
return | |
if "error" in result: | |
update.message.reply_text(f"Ошибка API: {result['error']}") | |
return | |
if not result.get("matches"): | |
update.message.reply_text("Совпадений не найдено.") | |
else: | |
# Отправляем совпадения пользователю | |
for match in result["matches"]: | |
url = match.get("url", "No URL provided") | |
img_base64 = match.get("image") | |
if img_base64: | |
# Конвертируем base64 обратно в изображение | |
img = Image.open(BytesIO(base64.b64decode(img_base64))) | |
bio = BytesIO() | |
img.save(bio, format="PNG") | |
bio.seek(0) | |
update.message.reply_photo(photo=InputFile(bio), caption=f"Найдено: {url}") | |
def main(): | |
"""Запускает бота.""" | |
token = "7547882359:AAEvPupQV2CdGgRrs_aqoYnjpHCj41baK7c" # Ваш токен Telegram | |
updater = Updater(token) | |
dispatcher = updater.dispatcher | |
# Регистрируем обработчики | |
dispatcher.add_handler(CommandHandler("start", start)) | |
dispatcher.add_handler(MessageHandler(Filters.photo, handle_image)) | |
# Запуск бота | |
updater.start_polling() | |
updater.idle() | |
if __name__ == "__main__": | |
main() |