Spaces:
Running
Running
File size: 4,010 Bytes
c7a5712 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
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() |