trueconf_bot/photo_bot/ai_validator.py

# /opt/trueconf_bot/photo_bot/ai_validator.py
# Тестовый мод фото-бота: прямая проверка фото ИИ (llama.cpp).
# Endpoint, модель и промпт — 1:1 из SibCemProfile/PhotoApp/photo_validator.py.
# Фото никуда не отправляется: только в ИИ, ответ возвращается пользователю.
import base64
import io
import json
import logging
import time
import urllib.request
from PIL import Image

import config.config as config

logger = logging.getLogger(__name__)

# Тег рассуждений thinking-моделей (конкатенация без литерала)
THINK_CLOSE_TAG = '</' + 'think>'


def _build_prompt() -> str:
    """Промпт проверки фото (без подсказки MediaPipe — в тест-моде геометрия не считается)."""
    cv_hint = 'ℹ️ ПОДСКАЗКА ОТ ALGORITHM: Геометрическая проверка в тестовом режиме не выполняется, проверяй визуально.'

    return ('''Ты — строгий инспектор пропускной системы. Проверь фотографию сотрудника на соответствие регламенту.

%s

ЗАПРЕЩЕНО выводить внутренние рассуждения и теги размышлений. Сразу пиши финальный ответ.

ПРЯМО СЕЙЧАС ВНИМАТЕЛЬНО РАССМОТРИ ФОТО И ПОСЛЕДОВАТЕЛЬНО ЗАФИКСИРУЙ:
A) РЕАЛЬНОСТЬ: это живое фото, а не ИИ-генерация, рисунок, 3D, мультфильм, фильтр?
B) ГРАФИКА: есть ли рамка, любые буквы/надписи/подписи (включая подпись с именем внизу), водяные знаки, декоративный фон с буквами, коллаж?
C) МИМИКА: высунут ли язык? Скривлены ли губы? Есть ли гримаса, смешная рожа, щёлканье зубами, смех, скос глаз, закатанные глаза?
D) РАКУРС: профиль, затылок, взгляд в сторону?
E) МАСШТАБ: ниже линии груди видны ли живот, талия, подмышки?

ПРАВИЛО ОТКАЗА: если по любому пункту A–E есть хоть малейшее отклонение — вердикт «Не прошел».
Смех, ржа, рожи, высунутый язык, скривлённые губы — фото НЕ ПРОХОДИТ, даже если лицо анфас и хорошо освещено.
Рамка или буквы/подпись на фоне или по краю кадра — фото НЕ ПРОХОДИТ.

ИТОГОВЫЙ ВЕРДИКТ:
- «Прошел» — только если все пункты A–E идеальны: нейтральное спокойное лицо, чистый кадр без рамки и букв.
- «Не прошел» — любое отклонение, описанное выше.
- «На модерацию» — только если фото реальное и лицо нейтральное, но масштаб или ракурс на грани нормы.

ФОРМАТ ОТВЕТА — СТРОГО 4 строки, без других слов:
Мимика: <нейтрально / высунут язык / гримаса / скривленная улыбка / смех — что именно видно>
Графика: <нет / рамка / буквы / подпись / коллаж — что именно видно>
Вердикт: [Прошел / Не прошел / На модерацию]
Причина: [1–2 предложения с указанием конкретного пункта A–E]
''') % cv_hint


def validate_photo_with_ai(file_bytes: bytes) -> dict:
    """
    Прямая проверка фото ИИ (тестовый мод).

    Возвращает {'ok': bool, 'verdict': str, 'reason': str, 'error': str,
                'metrics': {'total_time', 'prompt_tokens', 'completion_tokens',
                            'prompt_per_second', 'predicted_per_second'},
                'error_time': float}
    """
    t0 = time.perf_counter()
    try:
        # Сжатие (PIL): разрешение настраивается в config (PHOTO_TEST_AI_MAX_SIDE)
        side = int(config.PHOTO_TEST_AI_MAX_SIDE)
        with Image.open(io.BytesIO(file_bytes)) as img:
            img = img.convert('RGB')
            img.thumbnail((side, side))
            buffer = io.BytesIO()
            img.save(buffer, format='JPEG', quality=85)
            b64_img = base64.b64encode(buffer.getvalue()).decode('utf-8')

        payload = {
            'model': config.PHOTO_TEST_AI_MODEL,
            'messages': [{
                'role': 'user',
                'content': [
                    {'type': 'text', 'text': _build_prompt()},
                    {'type': 'image_url', 'image_url': {'url': 'data:image/jpeg;base64,' + b64_img}}
                ]
            }],
            'temperature': 0.1,
            'max_tokens': 1000
        }

        req = urllib.request.Request(
            config.PHOTO_TEST_AI_URL,
            data=json.dumps(payload).encode('utf-8'),
            headers={'Content-Type': 'application/json'}
        )

        with urllib.request.urlopen(req, timeout=config.PHOTO_TEST_AI_TIMEOUT) as resp:
            res = json.loads(resp.read().decode('utf-8'))
        total_time = time.perf_counter() - t0

        # llama.cpp в ответ кладёт usage (токены) и timings (скорости)
        usage = res.get('usage') or {}
        timings = res.get('timings') or {}
        metrics = {
            'total_time': total_time,
            'prompt_tokens': usage.get('prompt_tokens'),
            'completion_tokens': usage.get('completion_tokens'),
            'prompt_per_second': timings.get('prompt_per_second'),
            'predicted_per_second': timings.get('predicted_per_second'),
        }

        full_content = (res['choices'][0]['message'].get('content') or '').strip()
        clean_ans = full_content.split(THINK_CLOSE_TAG)[-1].strip() if THINK_CLOSE_TAG in full_content else full_content.strip()

        text_lower = clean_ans.lower()
        if 'вердикт: прошел' in text_lower or ('прошел' in text_lower and 'не прошел' not in text_lower and 'модерац' not in text_lower):
            verdict = 'Прошел'
        elif 'на модерацию' in text_lower or 'модерац' in text_lower:
            verdict = 'На модерацию'
        else:
            verdict = 'Не прошел'

        reason = ''
        mimi = ''
        grafika = ''
        for line in clean_ans.splitlines():
            ls = line.strip().lower()
            if ls.startswith('причина:'):
                reason = line.split(':', 1)[1].strip()
            elif ls.startswith('мимика:'):
                mimi = line.split(':', 1)[1].strip()
            elif ls.startswith('графика:'):
                grafika = line.split(':', 1)[1].strip()
        if not reason:
            reason = clean_ans.replace('\n', ' ').strip()

        logger.info(
            f"[TEST MODE] ИИ-вердикт: {verdict} | мимика={mimi} | графика={grafika} | {reason}"
            f" | total={metrics['total_time']:.2f}s "
            f"prompt={metrics['prompt_tokens']}tok/({metrics['prompt_per_second']:.1f} tps) "
            f"gen={metrics['completion_tokens']}tok/({metrics['predicted_per_second']:.1f} tps)"
        )
        return {'ok': True, 'verdict': verdict, 'reason': reason, 'error': '',
                'mimi': mimi, 'grafika': grafika, 'metrics': metrics}

    except Exception as e:
        err = f"{type(e).__name__}: {e}"
        elapsed = time.perf_counter() - t0
        logger.error(f"[TEST MODE] Ошибка обращения к ИИ через {elapsed:.1f} сек: {err}")
        return {'ok': False, 'verdict': '', 'reason': '', 'error': err,
                'error_time': elapsed}