294 lines
10 KiB
Python
294 lines
10 KiB
Python
from database.repository import (
|
||
find_instructions,
|
||
get_instruction_id_by_title,
|
||
find_tech_problems,
|
||
get_tech_solution,
|
||
get_tech_problem_by_name,
|
||
find_tech_solutions_by_name_contains,
|
||
set_tech_problem_progress,
|
||
get_tech_problem_progress,
|
||
clear_tech_problem_progress,
|
||
)
|
||
from services.instructions import (
|
||
send_pinpad_error,
|
||
send_terminal_instruction,
|
||
)
|
||
from keyboards.factory import back_to_main, build_keyboard, tech_feedback_keyboard
|
||
|
||
_DOOR_MAIN_LABELS = {"дверь", "двери"}
|
||
_DOOR_CATEGORY_LABELS = {
|
||
"Дверца накопителя",
|
||
"Входная дверь",
|
||
"Рекламная дверь",
|
||
"Дверь кассовой зоны",
|
||
}
|
||
|
||
_DOOR_ACCUMULATOR_QUERIES = ["дверца накопителя"]
|
||
_DOOR_ENTRANCE_QUERIES = ["входная дверь", "входной двери", "ключ от входной двери"]
|
||
_DOOR_ADS_QUERIES = ["рекламная дверь"]
|
||
_DOOR_KZ_QUERIES = ["дверь кз", "дверь кассовой зоны"]
|
||
_DOOR_ACCUMULATOR_REQUIRED = ["накопител"]
|
||
_DOOR_ENTRANCE_REQUIRED = ["входн"]
|
||
_DOOR_ADS_REQUIRED = ["рекламн"]
|
||
_DOOR_KZ_REQUIRED = ["кз", "кассов"]
|
||
|
||
|
||
async def handle_fallback(message):
|
||
text = (message.text or "").strip().lower()
|
||
|
||
if not text:
|
||
await message.answer(
|
||
"✍️ Введите код ошибки или опишите проблему.",
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
if message.text in _DOOR_CATEGORY_LABELS:
|
||
if message.text == "Дверца накопителя":
|
||
await _send_door_submenu(
|
||
message,
|
||
_DOOR_ACCUMULATOR_QUERIES,
|
||
"Выберите проблему с дверцей накопителя:",
|
||
required_substrings=_DOOR_ACCUMULATOR_REQUIRED,
|
||
)
|
||
return
|
||
if message.text == "Входная дверь":
|
||
await _send_door_submenu(
|
||
message,
|
||
_DOOR_ENTRANCE_QUERIES,
|
||
"Выберите проблему с входной дверью:",
|
||
required_substrings=_DOOR_ENTRANCE_REQUIRED,
|
||
)
|
||
return
|
||
if message.text == "Рекламная дверь":
|
||
await _send_door_submenu(
|
||
message,
|
||
_DOOR_ADS_QUERIES,
|
||
"Выберите проблему с рекламной дверью:",
|
||
required_substrings=_DOOR_ADS_REQUIRED,
|
||
)
|
||
return
|
||
if message.text == "Дверь кассовой зоны":
|
||
await _send_door_submenu(
|
||
message,
|
||
_DOOR_KZ_QUERIES,
|
||
"Выберите проблему с дверью кассовой зоны:",
|
||
required_substrings=_DOOR_KZ_REQUIRED,
|
||
)
|
||
return
|
||
|
||
# tech problem selection by name (from list)
|
||
tech_choice = get_tech_problem_by_name(text)
|
||
if tech_choice:
|
||
problem_id, _problem_name, task_type = tech_choice
|
||
await _send_tech_solution(message, problem_id, task_type)
|
||
return
|
||
|
||
# door уточнение
|
||
if text in _DOOR_MAIN_LABELS or any(part.startswith("двер") for part in text.split()):
|
||
keyboard = build_keyboard(
|
||
[{"title": t} for t in _DOOR_CATEGORY_LABELS],
|
||
back=True,
|
||
)
|
||
await message.answer(
|
||
"🚪 Уточните, какая дверь:",
|
||
keyboard=keyboard,
|
||
)
|
||
return
|
||
|
||
# tech problem feedback (text fallback)
|
||
if text in ("да", "нет"):
|
||
progress = get_tech_problem_progress(message.from_id)
|
||
if progress:
|
||
await _handle_tech_feedback(
|
||
user_id=message.from_id,
|
||
is_yes=(text == "да"),
|
||
send_fn=lambda msg: message.answer(msg, keyboard=back_to_main()),
|
||
)
|
||
return
|
||
|
||
# terminal instruction selection by title
|
||
instruction_id = get_instruction_id_by_title(text)
|
||
if instruction_id:
|
||
await send_terminal_instruction(
|
||
message,
|
||
instruction_id,
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
# 1️⃣ PINPAD — быстрый путь
|
||
if text.isdigit():
|
||
handled = await send_pinpad_error(
|
||
message,
|
||
text,
|
||
keyboard=back_to_main(),
|
||
)
|
||
if handled:
|
||
return
|
||
|
||
# 2️⃣ TERMINAL — поиск инструкций (приоритет для кассы/эвотор/чек)
|
||
found = find_instructions(text)
|
||
|
||
if not found:
|
||
# fall through to tech problems
|
||
found = []
|
||
|
||
if len(found) == 1:
|
||
instruction_id, _ = found[0]
|
||
await send_terminal_instruction(
|
||
message,
|
||
instruction_id,
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
if len(found) > 1:
|
||
buttons = [
|
||
{"title": title, "instruction_id": iid}
|
||
for iid, title in found
|
||
]
|
||
|
||
keyboard = build_keyboard(buttons, back=True)
|
||
|
||
await message.answer(
|
||
"🔎 Найдено несколько вариантов. Выберите нужный:",
|
||
keyboard=keyboard,
|
||
)
|
||
return
|
||
|
||
# 3️⃣ TECH PROBLEMS — поиск по тех. проблемам
|
||
# (дойти сюда можно только если терминал ничего не нашёл)
|
||
tech_found = find_tech_problems(text)
|
||
if tech_found:
|
||
if len(tech_found) > 1:
|
||
buttons = []
|
||
for pid, _task_type in tech_found[:6]:
|
||
solution = get_tech_solution(pid)
|
||
if solution:
|
||
buttons.append({"title": solution[1]})
|
||
if buttons:
|
||
keyboard = build_keyboard(buttons, back=True)
|
||
await message.answer(
|
||
"🔎 Найдено несколько тех. проблем. Выберите нужную:",
|
||
keyboard=keyboard,
|
||
)
|
||
return
|
||
|
||
problem_id, task_type = tech_found[0]
|
||
await _send_tech_solution(message, problem_id, task_type)
|
||
return
|
||
|
||
await message.answer(
|
||
"❌ Не удалось найти подходящую инструкцию.\n"
|
||
"✍️ Попробуйте изменить формулировку.",
|
||
keyboard=back_to_main(),
|
||
)
|
||
|
||
|
||
async def _send_tech_solution(message, problem_id: str, task_type: str):
|
||
solution = get_tech_solution(problem_id)
|
||
if not solution:
|
||
await message.answer(
|
||
("📝 Заполните форму:\nhttps://example.com/admin-form"
|
||
if task_type == "ADMIN"
|
||
else "📝 Заполните форму:\nhttps://example.com/tech-form"),
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
(
|
||
_pid, problem_name, task_type, can_fix_self,
|
||
need_result_feedback, solution_steps, tools_needed,
|
||
when_stop_and_report,
|
||
) = solution
|
||
|
||
if can_fix_self == "NO":
|
||
await message.answer(
|
||
f"⚠️ {problem_name}\n"
|
||
"Самостоятельно не исправляется.\n"
|
||
+ ("📝 Форма:\nhttps://example.com/admin-form"
|
||
if task_type == "ADMIN"
|
||
else "📝 Форма:\nhttps://example.com/tech-form"),
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
msg = f"🛠 {problem_name}\n\n✅ Шаги:\n{solution_steps}"
|
||
if tools_needed and tools_needed not in ("—", "-"):
|
||
msg += f"\n\n🧰 Инструменты:\n{tools_needed}"
|
||
if when_stop_and_report:
|
||
msg += f"\n\n⛔ Когда остановиться и сообщить:\n{when_stop_and_report}"
|
||
|
||
await message.answer(msg, keyboard=back_to_main())
|
||
|
||
if need_result_feedback == "YES":
|
||
set_tech_problem_progress(message.from_id, problem_id, task_type, need_result_feedback)
|
||
await message.answer("❓ Помогло?", keyboard=tech_feedback_keyboard())
|
||
|
||
|
||
def _name_has_required(name: str, required_substrings):
|
||
if not required_substrings:
|
||
return True
|
||
name_l = (name or "").lower()
|
||
return any(sub in name_l for sub in required_substrings)
|
||
|
||
|
||
async def _send_door_submenu(message, queries, prompt: str, required_substrings=None):
|
||
uniq = {}
|
||
for q in queries:
|
||
# by solution name
|
||
for pid, name, task_type in find_tech_solutions_by_name_contains(q):
|
||
if _name_has_required(name, required_substrings):
|
||
uniq[pid] = (name, task_type)
|
||
# by keywords (TechProblems)
|
||
for pid, _task_type in find_tech_problems(q):
|
||
solution = get_tech_solution(pid)
|
||
if solution:
|
||
if _name_has_required(solution[1], required_substrings):
|
||
uniq[pid] = (solution[1], solution[2])
|
||
|
||
if not uniq:
|
||
await message.answer(
|
||
"❌ Не удалось найти варианты по этой категории.",
|
||
keyboard=back_to_main(),
|
||
)
|
||
return
|
||
|
||
if len(uniq) == 1:
|
||
pid, (name, task_type) = next(iter(uniq.items()))
|
||
await _send_tech_solution(message, pid, task_type)
|
||
return
|
||
|
||
buttons = [{"title": name} for name, _t in list(uniq.values())[:6]]
|
||
keyboard = build_keyboard(buttons, back=True)
|
||
await message.answer(prompt, keyboard=keyboard)
|
||
|
||
|
||
async def _handle_tech_feedback(user_id: int, is_yes: bool, send_fn):
|
||
progress = get_tech_problem_progress(user_id)
|
||
if not progress:
|
||
return
|
||
|
||
_problem_id, task_type, _need_result_feedback = progress
|
||
clear_tech_problem_progress(user_id)
|
||
if is_yes:
|
||
await send_fn("✅ Отлично, рад помочь!")
|
||
return
|
||
|
||
form_url = "https://example.com/admin-form" if task_type == "ADMIN" else "https://example.com/tech-form"
|
||
await send_fn(f"📝 Заполните форму:\n{form_url}")
|
||
|
||
|
||
async def handle_tech_feedback_event(event) -> bool:
|
||
payload = event.get_payload_json() if hasattr(event, "get_payload_json") else None
|
||
if not payload or payload.get("tech_feedback") not in ("yes", "no"):
|
||
return False
|
||
|
||
await _handle_tech_feedback(
|
||
user_id=event.user_id,
|
||
is_yes=(payload.get("tech_feedback") == "yes"),
|
||
send_fn=lambda msg: event.send_message(msg, keyboard=back_to_main()),
|
||
)
|
||
return True
|