photo_bot: тестовый мод — фото напрямую в ИИ, без публикации

This commit is contained in:
Denis Krivochenko
2026-08-17 16:03:04 +07:00
parent 9f35db6f09
commit aee2fe8f25
5 changed files with 268 additions and 16 deletions
+15
View File
@@ -88,6 +88,21 @@ ENABLE_SEARCH_BOT = True # Модуль интеллектуальног
ENABLE_INSTRUCT = True # Модуль инструкций (Trueconf, Почта) ENABLE_INSTRUCT = True # Модуль инструкций (Trueconf, Почта)
ENABLE_LK = True # Личный кабинет сотрудника (отпуска, расчетные листки, справки) ENABLE_LK = True # Личный кабинет сотрудника (отпуска, расчетные листки, справки)
# =========================================================
# 1.1 ТЕСТОВЫЙ МОД ФОТО-БОТА
# Фото уходит НАПРЯМУЮ в ИИ (llama.cpp, тот же endpoint и промпт,
# что в SibCemProfile/PhotoApp/photo_validator.py), результат
# показывается пользователю. Само фото никуда НЕ ДОСЫЛАЕТСЯ
# не применяется (ни в AD, ни в API, ни на модерацию).
# =========================================================
PHOTO_TEST_MODE = os.getenv("PHOTO_TEST_MODE", "0").strip().lower() in ("1", "true", "yes", "on")
PHOTO_TEST_AI_URL = os.getenv("PHOTO_TEST_AI_URL", "http://srvkem-ii-03:8083/v1/chat/completions")
PHOTO_TEST_AI_MODEL = os.getenv(
"PHOTO_TEST_AI_MODEL",
"/opt/llama.cpp-cpu/build/bin/models/gemma-4-E4B-mtp/gemma-4-E4B-it-UD-Q4_K_XL.gguf"
)
PHOTO_TEST_AI_TIMEOUT = float(os.getenv("PHOTO_TEST_AI_TIMEOUT", "180"))
# ========================================================= # =========================================================
# 2. КОНТРОЛЬ ДОСТУПА # 2. КОНТРОЛЬ ДОСТУПА
# ========================================================= # =========================================================
+7 -2
View File
@@ -125,6 +125,7 @@ async def main_menu(msg: Message):
USER_MENU_SHOWN[user_id] = True USER_MENU_SHOWN[user_id] = True
from photo_bot.handlers import check_moderation_status from photo_bot.handlers import check_moderation_status
from utils.texts import photo_moderation_pending from utils.texts import photo_moderation_pending
if not PHOTO_TEST_MODE:
status = await check_moderation_status(user_id) status = await check_moderation_status(user_id)
if status.get("status") == "ok" and status.get("moderation_status") == "pending": if status.get("status") == "ok" and status.get("moderation_status") == "pending":
photo_id = status.get("photo_id", "???") photo_id = status.get("photo_id", "???")
@@ -135,8 +136,12 @@ async def main_menu(msg: Message):
) )
return return
set_state(user_id, "PHOTO_MODE") set_state(user_id, "PHOTO_MODE")
log_menu_stats(user_id, "Photo Bot", "Вход") log_menu_stats(user_id, "Photo Bot (TEST)" if PHOTO_TEST_MODE else "Photo Bot", "Вход")
await msg.answer(PHOTO_MAIN_MENU_TEXT, parse_mode="html") from utils.texts import PHOTO_TEST_WELCOME_TEXT
await msg.answer(
PHOTO_TEST_WELCOME_TEXT if PHOTO_TEST_MODE else PHOTO_MAIN_MENU_TEXT,
parse_mode="html"
)
return return
elif action == "TRANSCRIPTION": elif action == "TRANSCRIPTION":
USER_MENU_SHOWN[user_id] = True USER_MENU_SHOWN[user_id] = True
+113
View File
@@ -0,0 +1,113 @@
# /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 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
ЗАПРЕЩЕНО выводить внутренние рассуждения и теги размышлений. Сразу пиши финальный ответ.
БЛОК 0. КРИТИЧЕСКИЕ СТОП-ФАКТОРЫ:
1. Не реальное фото: ИИ-генерация, 3D-арт, цифровой рисунок, мультфильм, стилизованный портрет, сильная обработка фильтрами.
2. Графика и текст: Любые надписи, водяные знаки, рамки.
3. Костюмы и атрибуты: Исторические костюмы, короны, маски, каски (кроме строительных).
4. Посторонние предметы: Оружие, предметы перед лицом.
5. Неправильный ракурс и взгляд: Строгий боковой профиль, затылок, человек смотрит совсем в сторону.
БЛОК 1. МАСШТАБ И КАДРИРОВАНИЕ:
- подбородок НИЖЕ середины кадра → тест пройден.
- «Не прошел», если видны талия, живот, подмышки, одежда ниже линии груди.
ИТОГОВЫЙ ВЕРДИКТ:
- «Прошел» — нормальные портретные фотографии.
- «Не прошел» — только явные нарушения.
- «На модерацию» — если фото качественное, но масштаб на грани.
ФОРМАТ ОТВЕТА — СТРОГО 2 строки:
Вердикт: [Прошел / Не прошел / На модерацию]
Причина: [1–2 предложения с указанием конкретной причины]
''') % cv_hint
def validate_photo_with_ai(file_bytes: bytes) -> dict:
"""
Прямая проверка фото ИИ (тестовый мод).
Возвращает {'ok': bool, 'verdict': str, 'reason': str, 'error': str}
"""
try:
# Сжатие как в photo_validator.py (PIL, без сторонних DLL)
with Image.open(io.BytesIO(file_bytes)) as img:
img = img.convert('RGB')
img.thumbnail((768, 768))
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'))
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 = ''
for line in clean_ans.splitlines():
if line.strip().lower().startswith('причина:'):
reason = line.split(':', 1)[1].strip()
break
if not reason:
reason = clean_ans.replace('\n', ' ').strip()
logger.info(f"[TEST MODE] ИИ-вердикт: {verdict} | {reason}")
return {'ok': True, 'verdict': verdict, 'reason': reason, 'error': ''}
except Exception as e:
logger.error(f"[TEST MODE] Ошибка обращения к ИИ: {e}")
return {'ok': False, 'verdict': '', 'reason': '', 'error': str(e)}
+62 -1
View File
@@ -15,6 +15,7 @@ import config.config as config
from utils import states from utils import states
from utils.menu import MENU_TEXT from utils.menu import MENU_TEXT
from photo_bot.photo_processor import prepare_ad_photo, get_crop_coordinates from photo_bot.photo_processor import prepare_ad_photo, get_crop_coordinates
from photo_bot.ai_validator import validate_photo_with_ai
from utils.texts import ( from utils.texts import (
EMOJI_DIGITS, EMOJI_DIGITS,
PHOTO_MAIN_MENU_TEXT, PHOTO_MAIN_MENU_TEXT,
@@ -29,6 +30,12 @@ from utils.texts import (
photo_cancel_text, photo_cancel_text,
photo_preview_ready, photo_preview_ready,
photo_moderation_pending, photo_moderation_pending,
PHOTO_TEST_WELCOME_TEXT,
PHOTO_TEST_PREVIEW_TEXT,
PHOTO_TEST_RECEIVING_TEXT,
PHOTO_TEST_AI_CHECKING_TEXT,
photo_test_result,
photo_test_ai_error,
) )
# 🔌 Импортируем централизованную функцию сбора статистики из main # 🔌 Импортируем централизованную функцию сбора статистики из main
@@ -175,7 +182,8 @@ async def photo_module_handler(msg: Message):
return return
# --- ПЕРВАЯ ПРОВЕРКА: если нет фото в pending_photos, проверяем статус модерации --- # --- ПЕРВАЯ ПРОВЕРКА: если нет фото в pending_photos, проверяем статус модерации ---
if user_id not in pending_photos and not is_attachment: # (в тестовом моде API не трогаем — фото никуда не публикуется)
if user_id not in pending_photos and not is_attachment and not config.PHOTO_TEST_MODE:
status = await check_moderation_status(user_id) status = await check_moderation_status(user_id)
if status.get("status") == "ok" and status.get("moderation_status") == "pending": if status.get("status") == "ok" and status.get("moderation_status") == "pending":
photo_id = status.get("photo_id", "???") photo_id = status.get("photo_id", "???")
@@ -188,6 +196,53 @@ async def photo_module_handler(msg: Message):
# Ошибка проверки — не блокируем, просто логируем # Ошибка проверки — не блокируем, просто логируем
logger.warning(f"Не удалось проверить статус модерации для {user_id}: {status.get('reason')}") logger.warning(f"Не удалось проверить статус модерации для {user_id}: {status.get('reason')}")
# --- ТЕСТОВЫЙ МОД: кнопка 1 отправляет фото НАПРЯМУЮ в ИИ, фото никуда не публикуется ---
if not is_attachment and user_id in pending_photos and config.PHOTO_TEST_MODE:
if cmd == "1":
photo_data = pending_photos[user_id]
original_path = photo_data['path']
await msg.answer(PHOTO_TEST_AI_CHECKING_TEXT, parse_mode="html")
try:
with open(original_path, 'rb') as f:
raw_photo = f.read()
ai_res = await asyncio.to_thread(validate_photo_with_ai, raw_photo)
except Exception as e:
logger.error(f"🔥 [TEST MODE] Ошибка чтения фото: {e}")
ai_res = {'ok': False, 'verdict': '', 'reason': '', 'error': str(e)}
if not ai_res['ok']:
# Фото остаётся в pending — можно повторить проверку (кнопка 1)
log_menu_stats(user_id, "Photo Bot (TEST)", "Ошибка обращения к ИИ")
await msg.answer(photo_test_ai_error(), parse_mode="html")
return
verdict = ai_res['verdict']
reason = ai_res['reason'] or 'Причина не указана.'
log_menu_stats(user_id, "Photo Bot (TEST)", f"ИИ-вердикт: {verdict}")
# Фото остаётся в pending — можно повторить проверку другим фото (кнопка 2)
await msg.answer(photo_test_result(verdict, reason), parse_mode="html")
return
elif cmd == "2":
log_menu_stats(user_id, "Photo Bot (TEST)", "Загрузка другого фото")
photo_data = pending_photos[user_id]
original_path = photo_data['path']
if os.path.exists(original_path):
os.remove(original_path)
if 'preview_path' in photo_data and os.path.exists(photo_data['preview_path']):
os.remove(photo_data['preview_path'])
del pending_photos[user_id]
await msg.answer(PHOTO_TEST_WELCOME_TEXT, parse_mode="html")
return
# --- Защита от мусорного ввода при висящем превью (тестовый мод) ---
else:
await msg.answer(PHOTO_TEST_PREVIEW_TEXT, parse_mode="html")
return
# --- ПОДТВЕРЖДЕНИЕ ОТПРАВКИ (Кнопка 1) --- # --- ПОДТВЕРЖДЕНИЕ ОТПРАВКИ (Кнопка 1) ---
if not is_attachment and user_id in pending_photos: if not is_attachment and user_id in pending_photos:
if cmd == "1": if cmd == "1":
@@ -336,6 +391,12 @@ async def photo_module_handler(msg: Message):
log_menu_stats(user_id, "Photo Bot", "Загрузка и обработка нового фото") log_menu_stats(user_id, "Photo Bot", "Загрузка и обработка нового фото")
await asyncio.sleep(0.5) await asyncio.sleep(0.5)
if config.PHOTO_TEST_MODE:
await msg.answer(
PHOTO_TEST_PREVIEW_TEXT,
parse_mode="html"
)
else:
await msg.answer( await msg.answer(
photo_preview_ready(), photo_preview_ready(),
parse_mode="html" parse_mode="html"
+58
View File
@@ -243,6 +243,53 @@ def _photo_moderation_pending(photo_id: str, emojis: dict) -> str:
f"{emojis['0']}В главное меню" f"{emojis['0']}В главное меню"
) )
def _photo_test_welcome(emojis: dict) -> str:
return (
f"🧪 <b>Тестовый режим фото-проверки</b>\n\n"
f"📸 Пришлите фотографию — она уйдёт <b>напрямую в ИИ</b>.\n"
f"Бот покажет вердикт ИИ. <b>Фото никуда не публикуется</b>.\n\n"
f"{emojis['0']}В главное меню"
)
def _photo_test_preview(emojis: dict) -> str:
return (
f"🧪 <b>Тестовый режим</b>\n\n"
f"Кадр получен. Нажмите {emojis['1']} — отправлю фото напрямую в ИИ.\n"
f"<b>Фото никуда не публикуется.</b>\n\n"
f"{emojis['2']} — Загрузить другое фото\n"
f"{emojis['0']}В главное меню"
)
def _photo_test_ai_checking(emojis: dict) -> str:
return (
"🧪 <i>Отправляю фото напрямую в ИИ (тестовый режим)...\n"
"Обычно это занимает 1-3 минуты. Пожалуйста, подождите...</i>"
)
def _photo_test_receiving(emojis: dict) -> str:
return "🧪 <i>Ищу лицо на фото (тестовый режим), пожалуйста, подождите...</i>"
def _photo_test_result(verdict: str, reason: str, emojis: dict) -> str:
icons = {"Прошел": "", "Не прошел": "", "На модерацию": ""}
icon = icons.get(verdict, "")
return (
f"🧪 <b>Тестовый режим: вердикт ИИ</b>\n\n"
f"{icon} <b>Вердикт:</b> {verdict}\n"
f"📝 <b>Причина:</b> {reason}\n\n"
f"⚠️ <i>Тестовый режим: фото не публиковалось и никуда не отправлялось.</i>\n\n"
f"{emojis['2']} — Загрузить другое фото\n"
f"{emojis['0']}В главное меню"
)
def _photo_test_ai_error(emojis: dict) -> str:
return (
f"❌ <b>Ошибка обращения к ИИ.</b>\n\n"
f"Проверьте доступность ИИ-сервиса. Повторите попытку позже.\n\n"
f"{emojis['1']} — Повторить проверку\n"
f"{emojis['2']} — Загрузить другое фото\n"
f"{emojis['0']}В главное меню"
)
def _photo_success_published(emojis: dict) -> str: def _photo_success_published(emojis: dict) -> str:
return ( return (
f"🎉 <b>Фото успешно опубликовано!</b>\n\n" f"🎉 <b>Фото успешно опубликовано!</b>\n\n"
@@ -310,6 +357,17 @@ def photo_cancel_text() -> str:
def photo_moderation_pending(photo_id: str) -> str: def photo_moderation_pending(photo_id: str) -> str:
return _photo_moderation_pending(photo_id, EMOJI_DIGITS) return _photo_moderation_pending(photo_id, EMOJI_DIGITS)
PHOTO_TEST_WELCOME_TEXT = _photo_test_welcome(EMOJI_DIGITS)
PHOTO_TEST_PREVIEW_TEXT = _photo_test_preview(EMOJI_DIGITS)
PHOTO_TEST_AI_CHECKING_TEXT = _photo_test_ai_checking(EMOJI_DIGITS)
PHOTO_TEST_RECEIVING_TEXT = _photo_test_receiving(EMOJI_DIGITS)
def photo_test_result(verdict: str, reason: str) -> str:
return _photo_test_result(verdict, reason, EMOJI_DIGITS)
def photo_test_ai_error() -> str:
return _photo_test_ai_error(EMOJI_DIGITS)
# ========================================================= # =========================================================
# SEARCH BOT # SEARCH BOT