dddennnisss
612564389b
- make_face_crop_bytes(): тот же квадратный кроп по центру лица, что превью и публикуемый кадр (полное разрешение, JPEG q90) - handlers: кнопка 1 шлёт на проверку кроп, не оригинал - правило «голова >= 30%» теперь меряется по публикуемому результату
257 lines
10 KiB
Python
257 lines
10 KiB
Python
import io
|
|
import cv2
|
|
import logging
|
|
import numpy as np
|
|
import requests
|
|
from PIL import Image
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Настройки API
|
|
API_URL = "https://profile.sibcem.ru:8085/SibCemProfile/api/photo-bot/upload/"
|
|
API_KEY = "sibcem_bot_2026_a7f3k9m2x8q1w5e4r6t"
|
|
|
|
|
|
def _image_to_cv2(image_bytes: bytes):
|
|
"""Конвертирует байты изображения в OpenCV матрицу (поддержка webp)."""
|
|
# Сначала пробуем cv2.imdecode
|
|
nparr = np.frombuffer(image_bytes, np.uint8)
|
|
img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
|
|
if img is not None:
|
|
return img
|
|
|
|
# Если не сработало, пробуем через PIL (для webp и других форматов)
|
|
try:
|
|
pil_img = Image.open(io.BytesIO(image_bytes))
|
|
if pil_img.mode != 'RGB':
|
|
pil_img = pil_img.convert('RGB')
|
|
img = cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR)
|
|
return img
|
|
except Exception as e:
|
|
logger.error(f"Не удалось открыть изображение: {e}")
|
|
return None
|
|
|
|
|
|
def get_crop_coordinates(image_bytes: bytes) -> tuple | None:
|
|
"""
|
|
Ищет лицо на фото и возвращает координаты квадратного кропа.
|
|
|
|
Returns:
|
|
(crop_x1, crop_y1, crop_x2, crop_y2) или None если лицо не найдено.
|
|
"""
|
|
try:
|
|
img = _image_to_cv2(image_bytes)
|
|
if img is None:
|
|
logger.warning("Не удалось декодировать изображение")
|
|
return None
|
|
|
|
gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
|
|
|
|
cascade_path = cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'
|
|
face_cascade = cv2.CascadeClassifier(cascade_path)
|
|
faces = face_cascade.detectMultiScale(gray, scaleFactor=1.1, minNeighbors=8, minSize=(50, 50))
|
|
|
|
if len(faces) == 0:
|
|
logger.warning("Лицо не найдено на фото")
|
|
return None
|
|
|
|
if len(faces) > 1:
|
|
faces = sorted(faces, key=lambda f: f[2] * f[3], reverse=True)
|
|
|
|
x, y, w, h = faces[0]
|
|
|
|
margin = int(w * 0.3)
|
|
y1 = max(0, y - int(margin * 1.5))
|
|
y2 = min(img.shape[0], y + h + margin)
|
|
x1 = max(0, x - margin)
|
|
x2 = min(img.shape[1], x + w + margin)
|
|
|
|
side = min(x2 - x1, y2 - y1)
|
|
center_x, center_y = (x1 + x2) // 2, (y1 + y2) // 2
|
|
crop_x1 = max(0, center_x - side // 2)
|
|
crop_y1 = max(0, center_y - side // 2)
|
|
crop_x2 = crop_x1 + side
|
|
crop_y2 = crop_y1 + side
|
|
|
|
logger.info(f"Лицо найдено: ({x},{y},{w},{h}), кроп: ({crop_x1},{crop_y1},{crop_x2},{crop_y2})")
|
|
return (crop_x1, crop_y1, crop_x2, crop_y2)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Ошибка в get_crop_coordinates: {e}", exc_info=True)
|
|
return None
|
|
|
|
|
|
def _check_moderation_status(user_id: str) -> dict:
|
|
"""
|
|
Проверяет текущий статус модерации пользователя.
|
|
Возвращает {'status': 'ok', 'moderation_status': 'pending'|'published'|'none', ...}
|
|
"""
|
|
# Извлекаем логин из email (user_id@tcs.sibcem.ru -> user_id)
|
|
login = user_id.split('@')[0]
|
|
|
|
status_url = API_URL.replace("/upload/", "/status/")
|
|
try:
|
|
response = requests.get(
|
|
status_url,
|
|
params={"user_id": login},
|
|
headers={"X-API-Key": API_KEY},
|
|
timeout=30,
|
|
verify=False
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except Exception as e:
|
|
logger.error(f"Ошибка проверки статуса модерации: {e}")
|
|
return {"status": "error", "reason": f"Ошибка проверки статуса: {str(e)}"}
|
|
|
|
|
|
def _upload_to_api(image_bytes: bytes, user_id: str, coords: tuple) -> dict:
|
|
"""
|
|
Отправляет фото и координаты в Django API.
|
|
Возвращает JSON от сервера.
|
|
"""
|
|
x1, y1, x2, y2 = coords
|
|
try:
|
|
response = requests.post(
|
|
API_URL,
|
|
headers={"X-API-Key": API_KEY},
|
|
data={
|
|
"user_id": user_id,
|
|
"crop_x1": x1,
|
|
"crop_y1": y1,
|
|
"crop_x2": x2,
|
|
"crop_y2": y2,
|
|
},
|
|
files={
|
|
"image": ("photo.jpg", image_bytes, "image/jpeg")
|
|
},
|
|
timeout=300,
|
|
verify=False
|
|
)
|
|
response.raise_for_status()
|
|
# Логгируем ответ для отладки
|
|
logger.info(f"API response status: {response.status_code}, body[:500]: {response.text[:500]}")
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Ошибка API: {e}")
|
|
return {"status": "error", "reason": f"Ошибка сети: {str(e)}"}
|
|
except Exception as e:
|
|
logger.error(f"Ошибка отправки в API: {e}")
|
|
return {"status": "error", "reason": f"Ошибка: {str(e)}"}
|
|
|
|
|
|
def prepare_ad_photo(input_path: str, user_id: str, output_path: str = None, target_size: int = 96) -> dict:
|
|
"""
|
|
Отправляет фото на сервер через API.
|
|
output_path и target_size больше не используются (оставлены для обратной совместимости).
|
|
"""
|
|
try:
|
|
# Проверяем статус модерации перед отправкой
|
|
status = _check_moderation_status(user_id)
|
|
if status.get("status") == "ok" and status.get("moderation_status") == "pending":
|
|
photo_id = status.get("photo_id", "???")
|
|
return {
|
|
"success": False,
|
|
"error": f"У вас уже есть фото на модерации (ID: {photo_id}). Дождитесь решения модератора."
|
|
}
|
|
|
|
# 1. Читаем байты
|
|
if isinstance(input_path, bytes):
|
|
image_bytes = input_path
|
|
else:
|
|
with open(input_path, 'rb') as f:
|
|
image_bytes = f.read()
|
|
|
|
# 2. Ищем лицо и получаем координаты
|
|
coords = get_crop_coordinates(image_bytes)
|
|
if not coords:
|
|
return {"success": False, "error": "Лицо не найдено! Пожалуйста, сделайте фото анфас при хорошем освещении."}
|
|
|
|
# 3. Отправляем в API
|
|
api_result = _upload_to_api(image_bytes, user_id, coords)
|
|
|
|
# 4. Возвращаем результат
|
|
if api_result.get("status") == "ok":
|
|
return {
|
|
"success": True,
|
|
"photo_id": api_result.get("photo_id"),
|
|
"verdict": api_result.get("verdict"),
|
|
"message": api_result.get("message"),
|
|
}
|
|
else:
|
|
return {
|
|
"success": False,
|
|
"error": api_result.get("reason", "Неизвестная ошибка API")
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Ошибка в prepare_ad_photo: {e}", exc_info=True)
|
|
return {"success": False, "error": str(e)}
|
|
|
|
|
|
def make_face_crop_bytes(image_bytes: bytes, coords: tuple) -> bytes:
|
|
"""
|
|
Вырезает квадратный кадр по центру лица (тот же регион, что превью
|
|
бота и публикуемый на сайте кроп) и возвращает JPEG-байты.
|
|
|
|
На проверку уходит именно этот кроп: правило «голова ≥ 30%» и все
|
|
дальнейшие проверки имеют смысл для публикуемого результата, а не
|
|
для произвольно снятого оригинала.
|
|
|
|
Returns:
|
|
JPEG-байты квадрата (полное разрешение, без ресайза).
|
|
"""
|
|
from PIL import Image
|
|
import io
|
|
|
|
img = Image.open(io.BytesIO(image_bytes))
|
|
if img.mode != 'RGB':
|
|
img = img.convert('RGB')
|
|
|
|
x1, y1, x2, y2 = coords
|
|
cropped = img.crop((x1, y1, x2, y2))
|
|
|
|
# Квадрат из центра (как превью)
|
|
side = min(cropped.size[0], cropped.size[1])
|
|
left = (cropped.size[0] - side) // 2
|
|
top = (cropped.size[1] - side) // 2
|
|
cropped = cropped.crop((left, top, left + side, top + side))
|
|
|
|
buf = io.BytesIO()
|
|
cropped.save(buf, format="JPEG", quality=90)
|
|
return buf.getvalue()
|
|
|
|
|
|
def check_photo_api(image_bytes: bytes, timeout: float = 300.0) -> dict:
|
|
"""
|
|
Dry-run проверка фото через API сайта (POST /api/photo-bot/check/).
|
|
|
|
image_bytes — АВТО-КРОП ЛИЦА от бота (make_face_crop_bytes), тот же
|
|
кадр, что и превью/публикация. Фото проходит полный пайплайн проверки
|
|
на сайте (MediaPipe CV + ARKit-формы + VLM ИИ), НО ни в БД, ни в AD,
|
|
ни в Orion не попадает — ответ возвращается боту для тест-под-режима.
|
|
|
|
Returns:
|
|
{"status": "ok", "verdict", "ai_reason", "cv_approved", "cv_reason",
|
|
"raw_answer", "forced_moderation", "timings", "ai_metrics"}
|
|
или {"status": "error", "reason"}
|
|
"""
|
|
url = API_URL.replace("/upload/", "/check/")
|
|
try:
|
|
response = requests.post(
|
|
url,
|
|
headers={"X-API-Key": API_KEY},
|
|
files={"image": ("photo.jpg", image_bytes, "image/jpeg")},
|
|
timeout=timeout,
|
|
verify=False
|
|
)
|
|
response.raise_for_status()
|
|
logger.info(f"API /check/ response: status={response.status_code}, body[:500]: {response.text[:500]}")
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Ошибка /check/ API: {e}")
|
|
return {"status": "error", "reason": f"Ошибка сети при обращении к сайту: {e}"}
|
|
except Exception as e:
|
|
logger.error(f"Ошибка /check/: {e}")
|
|
return {"status": "error", "reason": f"Ошибка сайта проверки: {e}"}
|