pilot
/
restoran_max_bot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Добавление промо кода по запросу

Browse Source
main
pilot 2 months ago
parent bb166bbae3
commit 37007e6caf
11 changed files with 611 additions and 225 deletions
Show Stats Download Patch File Download Diff File

568
restoran_max_bot/ai.py
Unescape View File

@ -3,10 +3,25 @@ import json
import re
import shutil
import subprocess
import zipfile
from collections import defaultdict
from datetime import datetime
from pathlib import Path
# ---------- ЦВЕТА ДЛЯ КОНСОЛИ (НЕЙТРАЛЬНАЯ ПАЛИТРА) ----------
COLOR_RESET = "\033[0m"
COLOR_WHITE = "\033[97m" # белый (для заголовков)
COLOR_LIGHT_GRAY = "\033[37m" # светло-серый (для меню)
COLOR_GRAY = "\033[90m" # тёмно-серый (для информации)
COLOR_YELLOW = "\033[93m" # жёлтый (для ввода/предупреждений)
COLOR_GREEN = "\033[32m" # обычный зелёный (успех)
COLOR_RED = "\033[31m" # красный (ошибки)
COLOR_CYAN = "\033[36m" # голубой (акценты, откат)
COLOR_BOLD = "\033[1m"
def color_text(text, color=COLOR_RESET):
return f"{color}{text}{COLOR_RESET}"
# ---------- Функции сбора проекта ----------
def collect_project_files(root_dir, extensions=('.py', '.html'),
exclude_dirs=('venv', '__pycache__', '.git', 'env', 'node_modules')):
@ -146,11 +161,101 @@ def extract_json_from_text(text):
return None
# ---------- Применение изменений ----------
def apply_changes(project_root, actions):
# ---------- Полное резервное копирование проекта ----------
def create_full_backup(project_root, ai_response_dir):
"""Создаёт ZIP-архив всего проекта (исключая папку _ai_response)."""
backup_full_dir = os.path.join(ai_response_dir, 'backup', 'full')
os.makedirs(backup_full_dir, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
zip_filename = f"project_backup_{timestamp}.zip"
zip_path = os.path.join(backup_full_dir, zip_filename)
ignore_dir = os.path.basename(ai_response_dir)
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
for root, dirs, files in os.walk(project_root):
if ignore_dir in root.split(os.sep):
continue
for file in files:
file_path = os.path.join(root, file)
arcname = os.path.relpath(file_path, project_root)
zipf.write(file_path, arcname)
return zip_path
def list_full_backups(ai_response_dir):
backup_full_dir = os.path.join(ai_response_dir, 'backup', 'full')
if not os.path.exists(backup_full_dir):
return []
backups = []
for f in os.listdir(backup_full_dir):
if f.startswith('project_backup_') and f.endswith('.zip'):
full_path = os.path.join(backup_full_dir, f)
try:
time_str = f.replace('project_backup_', '').replace('.zip', '')
dt = datetime.strptime(time_str, "%Y%m%d_%H%M%S")
except:
dt = datetime.fromtimestamp(os.path.getmtime(full_path))
backups.append((full_path, dt, f))
backups.sort(key=lambda x: x[1], reverse=True)
return backups
def restore_full_backup(project_root, backup_path):
for item in os.listdir(project_root):
if item == os.path.basename(os.path.join(project_root, '_ai_response')):
continue
item_path = os.path.join(project_root, item)
if os.path.isfile(item_path):
os.remove(item_path)
else:
shutil.rmtree(item_path)
with zipfile.ZipFile(backup_path, 'r') as zipf:
zipf.extractall(project_root)
return True
# ---------- Бэкапирование отдельных файлов (для отката) ----------
def backup_file(abs_path, ai_response_dir):
backup_root = os.path.join(ai_response_dir, 'backup')
os.makedirs(backup_root, exist_ok=True)
project_root = os.getcwd()
rel_path = os.path.relpath(abs_path, project_root)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
backup_rel = f"{rel_path}.{timestamp}.bak"
backup_abs = os.path.join(backup_root, backup_rel)
os.makedirs(os.path.dirname(backup_abs), exist_ok=True)
shutil.copy2(abs_path, backup_abs)
return backup_abs
# ---------- Применение изменений (с бэкапами и полным архивом) ----------
def apply_changes(project_root, actions, ai_response_dir, user_prompt=''):
reports = []
errors = False
project_root = os.path.abspath(project_root)
# Добавляем запрос пользователя в отчёт
if user_prompt:
reports.append(f"📝 Запрос пользователя: {user_prompt}")
# Создаём полный бэкап проекта
try:
backup_zip = create_full_backup(project_root, ai_response_dir)
reports.append(f"📦 Полный бэкап проекта сохранён: {backup_zip}")
except Exception as e:
reports.append(f"❌ Ошибка при создании полного бэкапа: {e}")
errors = True
# Сохраняем данные о применении (включая запрос)
last_apply_file = os.path.join(ai_response_dir, 'last_apply.json')
apply_data = {
'user_prompt': user_prompt,
'actions': actions,
'timestamp': datetime.now().isoformat()
}
with open(last_apply_file, 'w', encoding='utf-8') as f:
json.dump(apply_data, f, ensure_ascii=False, indent=2)
for action in actions:
action_type = action.get('action')
file_path = action.get('file_path')
@ -176,11 +281,14 @@ def apply_changes(project_root, actions):
if not os.path.exists(abs_path):
reports.append(f"⚠️ Файл {file_path} не существует, создаём новый (как create)")
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
else:
backup_file(abs_path, ai_response_dir)
with open(abs_path, 'w', encoding='utf-8') as f:
f.write(content)
reports.append(f"✅ Обновлён: {file_path} — {description}")
elif action_type == 'delete':
if os.path.exists(abs_path):
backup_file(abs_path, ai_response_dir)
os.remove(abs_path)
reports.append(f"✅ Удалён: {file_path} — {description}")
else:
@ -194,6 +302,85 @@ def apply_changes(project_root, actions):
return reports, errors
# ---------- Откат изменений (только последнего набора) ----------
def rollback_last_apply(project_root, ai_response_dir):
reports = []
errors = False
last_apply_file = os.path.join(ai_response_dir, 'last_apply.json')
if not os.path.exists(last_apply_file):
reports.append("❌ Нет сохранённых действий для отката (файл last_apply.json не найден).")
return reports, True
try:
with open(last_apply_file, 'r', encoding='utf-8') as f:
data = json.load(f)
actions = data.get('actions', [])
except Exception as e:
reports.append(f"❌ Ошибка чтения last_apply.json: {e}")
return reports, True
for action in reversed(actions):
action_type = action.get('action')
file_path = action.get('file_path')
description = action.get('description', '')
if not action_type or not file_path:
continue
norm_path = os.path.normpath(file_path.replace('/', os.sep))
abs_path = os.path.join(project_root, norm_path)
if not os.path.abspath(abs_path).startswith(project_root):
reports.append(f"❌ Опасный путь: {file_path} — выход за пределы проекта")
errors = True
continue
try:
if action_type == 'create':
if os.path.exists(abs_path):
os.remove(abs_path)
reports.append(f"🗑️ Удалён (откат create): {file_path} — {description}")
else:
reports.append(f"⚠️ Файл {file_path} не найден, пропускаем удаление (откат create)")
elif action_type == 'update':
backup_root = os.path.join(ai_response_dir, 'backup')
backup_files = []
for root, dirs, files in os.walk(backup_root):
for f in files:
if f.startswith(norm_path + '.') and f.endswith('.bak'):
full = os.path.join(root, f)
backup_files.append((full, os.path.getmtime(full)))
if backup_files:
backup_files.sort(key=lambda x: x[1], reverse=True)
latest_backup = backup_files[0][0]
shutil.copy2(latest_backup, abs_path)
reports.append(f"♻️ Восстановлен (откат update): {file_path} — из {os.path.basename(latest_backup)}")
else:
reports.append(f"❌ Не найден бэкап для {file_path}, пропускаем (откат update)")
errors = True
elif action_type == 'delete':
backup_root = os.path.join(ai_response_dir, 'backup')
backup_files = []
for root, dirs, files in os.walk(backup_root):
for f in files:
if f.startswith(norm_path + '.') and f.endswith('.bak'):
full = os.path.join(root, f)
backup_files.append((full, os.path.getmtime(full)))
if backup_files:
backup_files.sort(key=lambda x: x[1], reverse=True)
latest_backup = backup_files[0][0]
os.makedirs(os.path.dirname(abs_path), exist_ok=True)
shutil.copy2(latest_backup, abs_path)
reports.append(f"♻️ Восстановлен (откат delete): {file_path} — из {os.path.basename(latest_backup)}")
else:
reports.append(f"❌ Не найден бэкап для {file_path}, пропускаем (откат delete)")
errors = True
else:
reports.append(f"❌ Неизвестное действие '{action_type}' для {file_path} (откат)")
errors = True
except Exception as e:
reports.append(f"❌ Ошибка при откате {file_path}: {e}")
errors = True
return reports, errors
# ---------- Вспомогательные функции ----------
def get_downloads_folder():
home = Path.home()
@ -246,7 +433,7 @@ def run_tests(project_root, test_path=None):
return False, reports, str(e)
# ---------- Логирование в суточный файл (только для APPLY и кратких записей о тестах) ----------
# ---------- Логирование в суточный файл ----------
def write_log_entry(log_file_path, entry_type, lines, extra_info=''):
with open(log_file_path, 'a', encoding='utf-8') as f:
f.write("=" * 80 + "\n")
@ -260,155 +447,252 @@ def write_log_entry(log_file_path, entry_type, lines, extra_info=''):
f.write("=" * 80 + "\n\n")
# ---------- ОСНОВНАЯ ФУНКЦИЯ ----------
# ---------- ОСНОВНАЯ ФУНКЦИЯ (ЦИКЛ) ----------
def main():
print("=" * 60)
print(" Инструмент для работы с DeepSeek")
print("=" * 60)
mode = input(
"Выберите режим:\n"
" 1 — Создать промт (собрать проект и инструкцию)\n"
" 2 — Применить ответ DeepSeek (автоматически найти JSON в Downloads)\n"
" 3 — Запустить тесты (все или указать папку/файл)\n"
"По умолчанию (Enter) — режим 1\n"
"Ваш выбор: "
).strip()
project_root = os.getcwd()
ai_response_dir = os.path.join(project_root, '_ai_response')
os.makedirs(ai_response_dir, exist_ok=True)
# Суточный лог-файл (только для изменений и кратких записей о тестах)
today = datetime.now().strftime('%Y-%m-%d')
log_file = os.path.join(ai_response_dir, f'ai_log_{today}.txt')
print(color_text("=" * 60, COLOR_WHITE + COLOR_BOLD))
print(color_text(" Инструмент для работы с DeepSeek", COLOR_WHITE + COLOR_BOLD))
print(color_text("=" * 60, COLOR_WHITE + COLOR_BOLD))
while True:
print("\n" + color_text("Выберите режим:", COLOR_LIGHT_GRAY + COLOR_BOLD))
print(color_text(" 0 — Выход", COLOR_GRAY))
print(color_text(" 1 — Создать промт (собрать проект и инструкцию)", COLOR_LIGHT_GRAY))
print(color_text(" 2 — Применить ответ DeepSeek (автоматически найти JSON в Downloads)", COLOR_LIGHT_GRAY))
print(color_text(" 3 — Запустить тесты (все или указать папку/файл)", COLOR_LIGHT_GRAY))
print(color_text(" 4 — Откатить последние изменения (восстановить из бэкапов файлов)", COLOR_LIGHT_GRAY))
print(color_text(" 5 — Восстановить проект из полного бэкапа (архива)", COLOR_LIGHT_GRAY))
mode = input(color_text("Ваш выбор (0-5): ", COLOR_LIGHT_GRAY + COLOR_BOLD)).strip()
if mode == '0':
print(color_text("Выход. До свидания!", COLOR_GREEN))
break
if mode == '1':
print(color_text("\nСбор информации о проекте для отправки в DeepSeek.", COLOR_LIGHT_GRAY))
main_prompt = input(color_text("Введите ваш промт (одна строка, Enter для завершения): ", COLOR_YELLOW)).strip()
if not main_prompt:
print(color_text("Промт не может быть пустым. Завершение.", COLOR_RED))
continue
# Сохраняем запрос пользователя для последующего логирования
prompt_file = os.path.join(ai_response_dir, 'last_user_prompt.txt')
try:
with open(prompt_file, 'w', encoding='utf-8') as f:
f.write(main_prompt)
except Exception as e:
print(color_text(f"⚠️ Не удалось сохранить запрос: {e}", COLOR_YELLOW))
print(color_text(f"\nСканируем папку: {project_root}", COLOR_GRAY))
files_data = collect_project_files(project_root)
full_prompt = build_prompt(main_prompt, files_data)
output_file = os.path.join(ai_response_dir, "ai_promt.txt")
try:
with open(output_file, 'w', encoding='utf-8') as f:
f.write(full_prompt)
print(color_text(f"\n✅ Готово! Промт сохранён в файл: {output_file}", COLOR_GREEN))
print(color_text(f"Всего собрано файлов: {len(files_data)}", COLOR_GRAY))
print(color_text("\nТеперь скопируйте содержимое этого файла и отправьте в DeepSeek.", COLOR_YELLOW))
print(color_text("DeepSeek вернёт ответ с описанием и JSON-блоком.", COLOR_YELLOW))
print(color_text("Скопируйте весь ответ и сохраните в файл с именем deepseek_json_YYYYMMDD_HHMMSS.json в папке Downloads.", COLOR_YELLOW))
print(color_text("Затем запустите этот скрипт в режиме 2 для автоматического применения.", COLOR_YELLOW))
except Exception as e:
print(color_text(f"❌ Ошибка при сохранении: {e}", COLOR_RED))
elif mode == '2':
downloads_dir = get_downloads_folder()
json_file = find_latest_json_file(downloads_dir)
if json_file is None:
print(color_text(f"❌ Не найден JSON-файл с префиксом 'deepseek_json_' в папке {downloads_dir}", COLOR_RED))
continue
print(color_text(f"📂 Найден файл: {json_file}", COLOR_GRAY))
try:
with open(json_file, 'r', encoding='utf-8') as f:
text = f.read()
except Exception as e:
print(color_text(f"❌ Ошибка чтения файла: {e}", COLOR_RED))
continue
data = extract_json_from_text(text)
if data is None:
print(color_text("❌ Не удалось извлечь JSON из ответа. Убедитесь, что в ответе есть блок ```json ... ``` или валидный JSON.", COLOR_RED))
continue
actions = data.get('actions')
if actions is None:
print(color_text("❌ В JSON отсутствует поле 'actions'.", COLOR_RED))
continue
if not isinstance(actions, list):
print(color_text("❌ Поле 'actions' должно быть массивом.", COLOR_RED))
continue
if len(actions) == 0:
print(color_text("ℹ️ Нет действий для применения.", COLOR_YELLOW))
continue
# Читаем сохранённый запрос пользователя
user_prompt = ''
prompt_file = os.path.join(ai_response_dir, 'last_user_prompt.txt')
if os.path.exists(prompt_file):
try:
with open(prompt_file, 'r', encoding='utf-8') as f:
user_prompt = f.read().strip()
except Exception as e:
print(color_text(f"⚠️ Не удалось прочитать запрос пользователя: {e}", COLOR_YELLOW))
print(color_text(f"\n📁 Применяем изменения в: {project_root}", COLOR_LIGHT_GRAY))
reports, errors = apply_changes(project_root, actions, ai_response_dir, user_prompt)
dest_filename = f"applied_{os.path.basename(json_file)}"
dest_path = os.path.join(ai_response_dir, dest_filename)
shutil.move(json_file, dest_path)
move_info = f"JSON перемещён в {dest_path}"
today = datetime.now().strftime('%Y-%m-%d')
log_file = os.path.join(ai_response_dir, f'ai_log_{today}.txt')
write_log_entry(log_file, 'APPLY', reports, extra_info=move_info)
print(color_text("\n" + "=" * 60, COLOR_WHITE + COLOR_BOLD))
print(color_text("ОТЧЁТ О ВЫПОЛНЕННЫХ ДЕЙСТВИЯХ", COLOR_WHITE + COLOR_BOLD))
print(color_text("=" * 60, COLOR_WHITE + COLOR_BOLD))
for line in reports:
if line.startswith("✅"):
print(color_text(line, COLOR_GREEN))
elif line.startswith("❌"):
print(color_text(line, COLOR_RED))
elif line.startswith("⚠️"):
print(color_text(line, COLOR_YELLOW))
elif line.startswith("📦"):
print(color_text(line, COLOR_CYAN))
else:
print(line)
if mode == '2':
# ---- РЕЖИМ 2: Применение изменений (пишем в суточный лог) ----
downloads_dir = get_downloads_folder()
json_file = find_latest_json_file(downloads_dir)
if json_file is None:
print(f"❌ Не найден JSON-файл с префиксом 'deepseek_json_' в папке {downloads_dir}")
return
if errors:
print(color_text("\n❌ Были ошибки при применении изменений.", COLOR_RED))
else:
print(color_text("\n✅ Все изменения применены успешно.", COLOR_GREEN))
print(color_text(f"\n📄 Лог изменений записан в: {log_file}", COLOR_GRAY))
elif mode == '3':
test_path = input(color_text("Введите относительный путь к папке или файлу с тестами (оставьте пустым для всех тестов): ", COLOR_YELLOW)).strip()
if test_path == '':
test_path = None
print(color_text("\n🔍 Запуск тестов...", COLOR_LIGHT_GRAY))
success, reports, output = run_tests(project_root, test_path)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
test_report_file = os.path.join(ai_response_dir, f"test_report_{timestamp}.txt")
with open(test_report_file, 'w', encoding='utf-8') as f:
f.write("=" * 60 + "\n")
f.write("ОТЧЁТ О ТЕСТИРОВАНИИ\n")
f.write(f"Дата: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"Путь тестов: {test_path if test_path else 'все тесты'}\n")
f.write("=" * 60 + "\n")
f.write(output)
today = datetime.now().strftime('%Y-%m-%d')
log_file = os.path.join(ai_response_dir, f'ai_log_{today}.txt')
brief_lines = [
f"Тесты запущены для пути: {test_path if test_path else 'все тесты'}",
f"Результат: {'✅ УСПЕШНО' if success else '❌ ОШИБКА'}",
f"Полный отчёт сохранён в: {test_report_file}"
]
write_log_entry(log_file, 'TEST', brief_lines, extra_info=f"Путь: {test_path if test_path else 'все тесты'}")
print(color_text("\n" + "=" * 60, COLOR_WHITE + COLOR_BOLD))
print(color_text("РЕЗУЛЬТАТ ТЕСТИРОВАНИЯ", COLOR_WHITE + COLOR_BOLD))
print(color_text("=" * 60, COLOR_WHITE + COLOR_BOLD))
for line in reports:
if line.startswith("✅"):
print(color_text(line, COLOR_GREEN))
elif line.startswith("❌"):
print(color_text(line, COLOR_RED))
else:
print(line)
print(f"📂 Найден файл: {json_file}")
if success:
print(color_text("\n✅ Тесты прошли успешно.", COLOR_GREEN))
else:
print(color_text("\n❌ Тесты завершились с ошибками. Проверьте полный отчёт.", COLOR_RED))
print(color_text(f"\n📄 Полный отчёт сохранён в: {test_report_file}", COLOR_GRAY))
print(color_text(f"📄 Краткая запись добавлена в суточный лог: {log_file}", COLOR_GRAY))
elif mode == '4':
print(color_text("\n🔄 Откат последних применённых изменений...", COLOR_LIGHT_GRAY))
reports, errors = rollback_last_apply(project_root, ai_response_dir)
today = datetime.now().strftime('%Y-%m-%d')
log_file = os.path.join(ai_response_dir, f'ai_log_{today}.txt')
write_log_entry(log_file, 'ROLLBACK', reports, extra_info="Откат выполненных изменений")
print(color_text("\n" + "=" * 60, COLOR_WHITE + COLOR_BOLD))
print(color_text("ОТЧЁТ ОБ ОТКАТЕ", COLOR_WHITE + COLOR_BOLD))
print(color_text("=" * 60, COLOR_WHITE + COLOR_BOLD))
for line in reports:
if line.startswith("✅"):
print(color_text(line, COLOR_GREEN))
elif line.startswith("❌"):
print(color_text(line, COLOR_RED))
elif line.startswith("⚠️"):
print(color_text(line, COLOR_YELLOW))
elif line.startswith("🗑️") or line.startswith("♻️"):
print(color_text(line, COLOR_CYAN))
else:
print(line)
try:
with open(json_file, 'r', encoding='utf-8') as f:
text = f.read()
except Exception as e:
print(f"❌ Ошибка чтения файла: {e}")
return
data = extract_json_from_text(text)
if data is None:
print("❌ Не удалось извлечь JSON из ответа. Убедитесь, что в ответе есть блок ```json ... ``` или валидный JSON.")
return
actions = data.get('actions')
if actions is None:
print("❌ В JSON отсутствует поле 'actions'.")
return
if not isinstance(actions, list):
print("❌ Поле 'actions' должно быть массивом.")
return
if len(actions) == 0:
print("ℹ️ Нет действий для применения.")
return
print(f"\n📁 Применяем изменения в: {project_root}")
reports, errors = apply_changes(project_root, actions)
# Перемещаем JSON-файл в _ai_response
dest_filename = f"applied_{os.path.basename(json_file)}"
dest_path = os.path.join(ai_response_dir, dest_filename)
shutil.move(json_file, dest_path)
move_info = f"JSON перемещён в {dest_path}"
# Пишем в суточный лог
write_log_entry(log_file, 'APPLY', reports, extra_info=move_info)
# Вывод в консоль
print("\n" + "=" * 60)
print("ОТЧЁТ О ВЫПОЛНЕННЫХ ДЕЙСТВИЯХ")
print("=" * 60)
for line in reports:
print(line)
if errors:
print("\n❌ Были ошибки при применении изменений.")
else:
print("\n✅ Все изменения применены успешно.")
print(f"\n📄 Лог изменений записан в: {log_file}")
elif mode == '3':
# ---- РЕЖИМ 3: Тестирование (отдельный файл + краткая запись в суточный лог) ----
test_path = input("Введите относительный путь к папке или файлу с тестами (оставьте пустым для всех тестов): ").strip()
if test_path == '':
test_path = None
print("\n🔍 Запуск тестов...")
success, reports, output = run_tests(project_root, test_path)
# Сохраняем полный вывод в отдельный файл
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
test_report_file = os.path.join(ai_response_dir, f"test_report_{timestamp}.txt")
with open(test_report_file, 'w', encoding='utf-8') as f:
f.write("=" * 60 + "\n")
f.write("ОТЧЁТ О ТЕСТИРОВАНИИ\n")
f.write(f"Дата: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write(f"Путь тестов: {test_path if test_path else 'все тесты'}\n")
f.write("=" * 60 + "\n")
f.write(output)
# Пишем краткую запись в суточный лог
brief_lines = [
f"Тесты запущены для пути: {test_path if test_path else 'все тесты'}",
f"Результат: {'✅ УСПЕШНО' if success else '❌ ОШИБКА'}",
f"Полный отчёт сохранён в: {test_report_file}"
]
write_log_entry(log_file, 'TEST', brief_lines, extra_info=f"Путь: {test_path if test_path else 'все тесты'}")
# Вывод в консоль
print("\n" + "=" * 60)
print("РЕЗУЛЬТАТ ТЕСТИРОВАНИЯ")
print("=" * 60)
for line in reports:
print(line)
if errors:
print(color_text("\n❌ Были ошибки при откате.", COLOR_RED))
else:
print(color_text("\n✅ Откат выполнен успешно.", COLOR_GREEN))
print(color_text(f"\n📄 Лог отката записан в: {log_file}", COLOR_GRAY))
elif mode == '5':
backups = list_full_backups(ai_response_dir)
if not backups:
print(color_text("❌ Нет доступных полных бэкапов.", COLOR_RED))
continue
print(color_text("\nДоступные полные бэкапы:", COLOR_LIGHT_GRAY))
for idx, (path, dt, name) in enumerate(backups, 1):
print(color_text(f" {idx}. {name} ({dt.strftime('%Y-%m-%d %H:%M:%S')})", COLOR_GRAY))
choice = input(color_text("Выберите номер бэкапа для восстановления (или 0 для отмены): ", COLOR_YELLOW)).strip()
if choice == '0':
continue
try:
idx = int(choice) - 1
if idx < 0 or idx >= len(backups):
print(color_text("❌ Неверный номер.", COLOR_RED))
continue
backup_path = backups[idx][0]
except ValueError:
print(color_text("❌ Неверный ввод.", COLOR_RED))
continue
print(color_text(f"\n⚠️ Восстановление из бэкапа {os.path.basename(backup_path)} полностью заменит текущий проект.", COLOR_YELLOW))
confirm = input(color_text("Вы уверены? (введите 'да' для подтверждения): ", COLOR_YELLOW)).strip().lower()
if confirm != 'да':
print(color_text("Восстановление отменено.", COLOR_GRAY))
continue
try:
restore_full_backup(project_root, backup_path)
print(color_text(f"\n✅ Проект восстановлен из бэкапа: {os.path.basename(backup_path)}", COLOR_GREEN))
today = datetime.now().strftime('%Y-%m-%d')
log_file = os.path.join(ai_response_dir, f'ai_log_{today}.txt')
write_log_entry(log_file, 'RESTORE_FULL', [f"Восстановлен проект из {os.path.basename(backup_path)}"], extra_info="Полное восстановление")
except Exception as e:
print(color_text(f"❌ Ошибка при восстановлении: {e}", COLOR_RED))
if success:
print("\n✅ Тесты прошли успешно.")
else:
print("\n❌ Тесты завершились с ошибками. Проверьте полный отчёт.")
print(f"\n📄 Полный отчёт сохранён в: {test_report_file}")
print(f"📄 Краткая запись добавлена в суточный лог: {log_file}")
else:
# ---- РЕЖИМ 1: Создание промта (сохраняем в _ai_response) ----
print("\nСбор информации о проекте для отправки в DeepSeek.")
main_prompt = input("\nВведите ваш промт (одна строка, Enter для завершения): ").strip()
if not main_prompt:
print("Промт не может быть пустым. Завершение.")
return
print(f"\nСканируем папку: {project_root}")
files_data = collect_project_files(project_root)
full_prompt = build_prompt(main_prompt, files_data)
output_file = os.path.join(ai_response_dir, "ai_promt.txt")
try:
with open(output_file, 'w', encoding='utf-8') as f:
f.write(full_prompt)
print(f"\n✅ Готово! Промт сохранён в файл: {output_file}")
print(f"Всего собрано файлов: {len(files_data)}")
print("\nТеперь скопируйте содержимое этого файла и отправьте в DeepSeek.")
print("DeepSeek вернёт ответ с описанием и JSON-блоком.")
print("Скопируйте весь ответ и сохраните в файл с именем deepseek_json_YYYYMMDD_HHMMSS.json в папке Downloads.")
print("Затем запустите этот скрипт в режиме 2 для автоматического применения.")
except Exception as e:
print(f"❌ Ошибка при сохранении: {e}")
print(color_text("❌ Неверный выбор. Пожалуйста, введите число от 0 до 5.", COLOR_RED))
if __name__ == "__main__":
main()

3
restoran_max_bot/content_bot/urls.py
Unescape View File

@ -1,8 +1,9 @@
from django.urls import path
from content_bot.views import api_default_load
from content_bot.views import api_default_load, ui_menu_view
from max_bot.views import api_start_max_v1, iiko_webhook, iiko_send_message
urlpatterns = [
path('default-load', api_default_load),
path('ui/', ui_menu_view, name='ui_menu'),
]

8
restoran_max_bot/content_bot/views.py
Unescape View File

@ -1,6 +1,7 @@
from django.http import JsonResponse
from django.core.handlers.wsgi import WSGIRequest
from django.views.decorators.csrf import csrf_exempt
from django.shortcuts import render
from max_bot.models import Client, ProductCategory, Product
@ -9,3 +10,10 @@ from max_bot.models import Client, ProductCategory, Product
def api_default_load(request: WSGIRequest):
rt = {'success': True, 'data': ''}
return JsonResponse(rt, status=200)
def ui_menu_view(request):
"""Отображает главную страницу MAX UI с двумя кнопками."""
# В будущем здесь можно будет получать данные клиента из сессии или параметров запроса
# Сейчас просто рендерим шаблон
return render(request, 'max_bot/ui_menu.html')

7
restoran_max_bot/max_bot/views.py
Unescape View File

@ -8,7 +8,8 @@ from max_bot.models import Partner, Integration, Client, BonusTransaction, App,
from restoran_max_bot import settings
from restoran_max_bot.bot_started import bot_started
from restoran_max_bot.bot_stopped import bot_stopped
from restoran_max_bot.bot_message import bot_message, bot_promo
from restoran_max_bot.bot_message import bot_message
from restoran_max_bot.promo_handlers import bot_promo
from restoran_max_bot.common import common_get_contact, common_check_registration
from restoran_max_bot.settings import DEBUG
from restoran_max_bot.utils import is_json, has_key
@ -62,7 +63,6 @@ def api_decorator(func):
@csrf_exempt
@api_decorator
def api_start_max_v1(request, token, **kwargs):
# основной обработчик MAX-БОТ
data = json.loads(request.body.decode('utf-8'))
client = kwargs['client']
settings_max = kwargs['settings_max']
@ -83,9 +83,6 @@ def api_start_max_v1(request, token, **kwargs):
# start - запуск бота
if data['update_type'] == 'bot_started':
# https://dev.max.ru/docs/chatbots/bots-coding/prepare
# https://max.ru/id032606883607_2_bot?start=source_site
# роверка на промокод
if data.get("payload", None):
bot_promo(client=client, message=data, settings_max=settings_max)
if not common_check_registration(client=client, chat_id=data['chat_id']):

14
restoran_max_bot/restoran_max_bot/ai_handler.py
Unescape View File

@ -1,6 +1,7 @@
# ai_handler.py
from ai_agent.api_common import get_ai_agent, call_ai_agent
from max_bot.models import Client
from restoran_max_bot.user_logger import log_user_action
def process_with_ai(client: Client, chat_id: str, user_query: str, contact) -> dict:
@ -16,4 +17,17 @@ def process_with_ai(client: Client, chat_id: str, user_query: str, contact) -> d
user_query=user_query,
contact_name=real_name
)
# Логируем сам факт обращения к AI (детали будут добавлены в bot_message)
# Здесь мы можем залогировать, что AI был вызван, но основное логирование уже есть в bot_message
# Поэтому оставляем пустым или можно залогировать вызов
log_user_action(
client=client,
chat_id=chat_id,
action_type='ai_call',
user_query=user_query,
details={
'ai_agent': ai_agent.get('ai_agent'),
'contact_name': real_name
}
)
return response

90
restoran_max_bot/restoran_max_bot/bot_message.py
Unescape View File

@ -10,6 +10,7 @@ from restoran_max_bot.command_handlers import handle_standard_command, handle_ai
from restoran_max_bot.ai_handler import process_with_ai
from restoran_max_bot.settings import STATICFILES_DIRS, BASE_URL
from restoran_max_bot.utils import is_json
from restoran_max_bot.user_logger import log_user_action
def bot_message(client: Client, message: dict, settings_max: dict, message_type: str) -> bool:
@ -42,17 +43,54 @@ def bot_message(client: Client, message: dict, settings_max: dict, message_type:
send_menu(chat_id, token, client)
return True
# Логируем входящий запрос
log_user_action(
client=client,
chat_id=chat_id,
action_type='incoming_message',
user_query=user_query,
details={
'message_type': message_type,
'contact_name': contact.name,
'phone': contact.phone
}
)
# 4. Обработка стандартных команд (меню, бонусы и т.п.)
data = _get_command_data(client, message, message_type)
if data:
handled = handle_standard_command(client, chat_id, token, data, contact)
handled = handle_standard_command(client, chat_id, settings_max, data, contact, command_key=user_query)
if handled:
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command_handled',
command=user_query,
details={
'command_type': 'standard',
'contact_name': contact.name
}
)
return True
# 5. Если не стандартная команда — пробуем AI
response = process_with_ai(client, chat_id, user_query, contact)
if response:
handle_ai_response(client, chat_id, token, response, contact)
# Логируем ответ AI
log_user_action(
client=client,
chat_id=chat_id,
action_type='ai_response',
user_query=user_query,
intent=response.get('intent'),
entities=response.get('entities'),
response=response.get('message'),
details={
'status': response.get('status'),
'contact_name': contact.name
}
)
send_menu(chat_id, token, client)
return True
@ -93,53 +131,3 @@ def _get_command_data(client: Client, message: dict, message_type: str):
if not key:
return None
return common_get_data(client=client, key=str(key).lower())
# ==================== bot_promo ====================
def bot_promo(client: Client, message: dict, settings_max: dict) -> bool:
"""Обработчик промокодов (QR-коды)."""
chat_id = message['chat_id']
promo = message['payload'].lower()
messenger = 'max-bot'
client_id = client.id
data = common_get_data(client=client, key=promo)
if not data:
return True
promo_obj, created = PromoCode.objects.get_or_create(
promo=promo,
chat_id=chat_id,
client=client,
defaults={'messenger': messenger, 'used': False}
)
if not created:
if promo_obj.used:
send_text(chat_id, settings_max['token'], "⚠️ Этот промокод уже был использован.")
else:
send_text(chat_id, settings_max['token'], "ℹ️ QR-код уже был отправлен.")
return True
# Генерируем QR
base_url = f"{BASE_URL}/promo/status/"
url_promo = f"{base_url}{promo}/{chat_id}/{client_id}"
img = qrcode.make(url_promo)
file_name = f"{promo}-{chat_id}-{client_id}.png"
img.save(f"{STATICFILES_DIRS[0]}/client_qr/{file_name}")
qr_url = f"{BASE_URL}/static/client_qr/{file_name}"
send_image(chat_id, settings_max['token'], qr_url, 'Ваш QR-код.')
# Отправляем дополнительный контент
for item in data:
if item.img:
img_url = f"https://cdn.telefon-ip.ru/{item.img}?thumb=600"
send_image(chat_id, settings_max['token'], img_url, item.descr)
if item.title and not item.img:
send_text(chat_id, settings_max['token'], item.descr)
if item.url:
send_text(chat_id, settings_max['token'], item.url)
return True

23
restoran_max_bot/restoran_max_bot/bot_started.py
Unescape View File

@ -1,6 +1,7 @@
from max_bot.max_api import *
from max_bot.models import Client
from restoran_max_bot.common import common_get_data, common_get_contact, common_set_contact
from restoran_max_bot.user_logger import log_user_action
def bot_started(client: Client, message: dict, settings: dict):
@ -8,6 +9,17 @@ def bot_started(client: Client, message: dict, settings: dict):
name = message['user']['name']
user_id = message['user']['user_id']
data = common_get_data(client=client, key='start')
log_user_action(
client=client,
chat_id=chat_id,
action_type='bot_started',
details={
'user_name': name,
'user_id': user_id
}
)
if data:
for dt in data:
if dt.img:
@ -69,6 +81,17 @@ def check_registration(client: Client, chat_id: str, message: dict, settings: di
contact = common_set_contact(client=client, uid_client=chat_id, month=dt['data'])
maxbot_send_content(client=client, chat_id=chat_id, max_token=settings['token'],
message='system_registration')
# Логируем успешную регистрацию
log_user_action(
client=client,
chat_id=chat_id,
action_type='registration_completed',
details={
'name': contact.name,
'phone': contact.phone,
'birthday': f"{contact.field1}.{contact.field2}"
}
)
return contact
else:
dt = maxbot_send_content(client=client, chat_id=chat_id, max_token=settings['token'],

85
restoran_max_bot/restoran_max_bot/command_handlers.py
Unescape View File

@ -9,28 +9,68 @@ from restoran_max_bot.message_sender import send_text, send_image, send_menu, se
from restoran_max_bot.settings import STATICFILES_DIRS, BASE_URL
from restoran_max_bot.utils import is_json
from restoran_max_bot.bot_started import check_registration
from restoran_max_bot.promo_handlers import bot_promo
from restoran_max_bot.user_logger import log_user_action
def handle_standard_command(client: Client, chat_id: str, token: str, data, contact: Contact) -> bool:
"""Обрабатывает стандартные команды (меню, бонусы, отзывы, бронирование).
def handle_standard_command(client: Client, chat_id: str, settings_max: dict, data, contact: Contact, command_key: str = None) -> bool:
"""Обрабатывает стандартные команды (меню, бонусы, отзывы, бронирование, промо).
Возвращает True, если команда была обработана, иначе False."""
if not data:
return False
token = settings_max['token']
for item in data:
# Промокод (команда ##promo##)
if item.title == '##promo##' and command_key:
fake_message = {'chat_id': chat_id, 'payload': command_key}
result = bot_promo(client, fake_message, settings_max)
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command',
command='promo',
details={'promo_key': command_key}
)
# После обработки промо-кода отправляем меню
send_menu(chat_id, token, client)
return result
# Бонусы
if item.title == '##bonus##':
_handle_bonus(client, chat_id, token, contact)
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command',
command='bonus',
details={'contact_name': contact.name}
)
continue
# Отзыв
if item.title == '##feedback##':
send_feedback_buttons(chat_id, token, client)
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command',
command='feedback',
details={'contact_name': contact.name}
)
return True
# Бронирование
if item.title == '##booking##':
send_booking_people_buttons(chat_id, token, client)
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command',
command='booking',
details={'contact_name': contact.name}
)
return True
# Кнопка-ссылка
@ -38,6 +78,13 @@ def handle_standard_command(client: Client, chat_id: str, token: str, data, cont
data_json = json.loads(item.title)
if data_json.get('type') == 'linkbutton':
send_link_button(chat_id, token, item.descr, data_json['button'], item.url)
log_user_action(
client=client,
chat_id=chat_id,
action_type='standard_command',
command='linkbutton',
details={'button_text': data_json['button']}
)
continue
# Изображение
@ -158,6 +205,18 @@ def _handle_booking_intent(chat_id: str, token: str, client: Client, response: d
success = _send_booking_to_crm(client, date, time, people, phone, contact.name)
if success:
send_text(chat_id, token, f"✅ Бронирование на {date} в {time} на {people} чел. принято! Администратор свяжется с вами.")
log_user_action(
client=client,
chat_id=chat_id,
action_type='booking_completed',
details={
'date': date,
'time': time,
'people': people,
'phone': phone,
'contact_name': contact.name
}
)
else:
send_text(chat_id, token, "❌ Не удалось забронировать. Попробуйте позже.")
else:
@ -185,6 +244,18 @@ def _handle_contact_admin_intent(chat_id: str, token: str, client: Client, respo
success = _send_contact_to_admin(client, phone, contact_method, question, preferred_time, contact.name)
if success:
send_text(chat_id, token, f"✅ Ваш запрос передан администратору. Способ связи: {contact_method}. Скоро с вами свяжутся.")
log_user_action(
client=client,
chat_id=chat_id,
action_type='contact_admin_completed',
details={
'phone': phone,
'contact_method': contact_method,
'question': question,
'preferred_time': preferred_time,
'contact_name': contact.name
}
)
else:
send_text(chat_id, token, "❌ Не удалось отправить запрос. Попробуйте позже.")
else:
@ -210,6 +281,16 @@ def _handle_feedback_intent(chat_id: str, token: str, client: Client, response:
success = _send_feedback_to_crm(client, rating, feedback_text, contact.name, contact.phone)
if success:
send_text(chat_id, token, "✅ Спасибо за ваш отзыв! Мы учтём его.")
log_user_action(
client=client,
chat_id=chat_id,
action_type='feedback_completed',
details={
'rating': rating,
'feedback_text': feedback_text,
'contact_name': contact.name
}
)
else:
send_text(chat_id, token, "❌ Не удалось отправить отзыв. Попробуйте позже.")
else:

12
restoran_max_bot/restoran_max_bot/settings.py
Unescape View File

@ -32,7 +32,6 @@ if DEBUG:
else:
BASE_URL = "https://maxbot.telefon-ip.ru"
ALLOWED_HOSTS = ['*']
@ -194,16 +193,5 @@ LOGGING = {
'level': 'ERROR',
'propagate': False,
},
# Можно добавить логгеры для своих приложений, если нужно
# 'max_bot': {
# 'handlers': ['file'],
# 'level': 'ERROR',
# 'propagate': False,
# },
# 'ai_agent': {
# 'handlers': ['file'],
# 'level': 'ERROR',
# 'propagate': False,
# },
},
}

24
restoran_max_bot/restoran_max_bot/tests.py
Unescape View File

@ -4,7 +4,8 @@ import unittest
from unittest.mock import Mock, patch, MagicMock
from django.test import TestCase
from max_bot.models import Client, Contact, Product, ProductCategory, PromoCode
from restoran_max_bot.bot_message import bot_message, bot_promo, _get_attachments, _extract_query, _get_command_data
from restoran_max_bot.bot_message import bot_message, _get_attachments, _extract_query, _get_command_data
from restoran_max_bot.promo_handlers import bot_promo
from restoran_max_bot.message_sender import send_text, send_image, send_menu
from restoran_max_bot.command_handlers import handle_standard_command, handle_ai_response
from restoran_max_bot.ai_handler import process_with_ai
@ -192,11 +193,11 @@ class BotMessageTestCase(TestCase):
# ---------- Тесты bot_promo ----------
def test_bot_promo_new(self):
"""Создание нового промокода."""
with patch('restoran_max_bot.bot_message.common_get_data') as mock_get_data, \
patch('restoran_max_bot.bot_message.PromoCode.objects.get_or_create') as mock_get_or_create, \
patch('restoran_max_bot.bot_message.qrcode.make') as mock_qr, \
patch('restoran_max_bot.bot_message.send_image') as mock_send_image, \
patch('restoran_max_bot.bot_message.send_text') as mock_send_text:
with patch('restoran_max_bot.promo_handlers.common_get_data') as mock_get_data, \
patch('restoran_max_bot.promo_handlers.PromoCode.objects.get_or_create') as mock_get_or_create, \
patch('restoran_max_bot.promo_handlers.qrcode.make') as mock_qr, \
patch('restoran_max_bot.promo_handlers.send_image') as mock_send_image, \
patch('restoran_max_bot.promo_handlers.send_text') as mock_send_text:
mock_get_data.return_value = [Mock(img='', title='', url='')]
mock_get_or_create.return_value = (Mock(used=False), True) # created=True
mock_qr.return_value = Mock()
@ -207,7 +208,7 @@ class BotMessageTestCase(TestCase):
mock_send_image.assert_called_once()
mock_send_text.assert_not_called()
@patch('restoran_max_bot.bot_message.common_get_data')
@patch('restoran_max_bot.promo_handlers.common_get_data')
def test_bot_promo_no_data(self, mock_get_data):
"""Промокод не найден в системе."""
mock_get_data.return_value = None
@ -243,6 +244,7 @@ class CommandHandlersTestCase(TestCase):
)
self.chat_id = 'chat123'
self.token = 'token'
self.settings_max = {'token': self.token}
@patch('restoran_max_bot.command_handlers.send_menu')
@patch('restoran_max_bot.command_handlers.send_text')
@ -254,7 +256,7 @@ class CommandHandlersTestCase(TestCase):
mock_item.descr = ''
mock_item.url = ''
with patch('restoran_max_bot.command_handlers._handle_bonus') as mock_handle_bonus:
result = handle_standard_command(self.client, self.chat_id, self.token,
result = handle_standard_command(self.client, self.chat_id, self.settings_max,
[mock_item], self.contact)
self.assertTrue(result)
mock_handle_bonus.assert_called_once()
@ -266,7 +268,7 @@ class CommandHandlersTestCase(TestCase):
from restoran_max_bot.command_handlers import handle_standard_command
mock_item = Mock()
mock_item.title = '##feedback##'
result = handle_standard_command(self.client, self.chat_id, self.token,
result = handle_standard_command(self.client, self.chat_id, self.settings_max,
[mock_item], self.contact)
self.assertTrue(result)
mock_send_feedback.assert_called_once()
@ -278,7 +280,7 @@ class CommandHandlersTestCase(TestCase):
from restoran_max_bot.command_handlers import handle_standard_command
mock_item = Mock()
mock_item.title = '##booking##'
result = handle_standard_command(self.client, self.chat_id, self.token,
result = handle_standard_command(self.client, self.chat_id, self.settings_max,
[mock_item], self.contact)
self.assertTrue(result)
mock_send_booking.assert_called_once()
@ -292,7 +294,7 @@ class CommandHandlersTestCase(TestCase):
mock_item.title = json.dumps({'type': 'linkbutton', 'button': 'Click'})
mock_item.descr = 'Text'
mock_item.url = 'http://example.com'
result = handle_standard_command(self.client, self.chat_id, self.token,
result = handle_standard_command(self.client, self.chat_id, self.settings_max,
[mock_item], self.contact)
self.assertTrue(result)
mock_send_link.assert_called_once()

2
restoran_max_bot/templates/max_bot/promo_status.html
Unescape View File

@ -212,4 +212,4 @@
})();
</script>
</body>
</html>
</html>

Write Preview
Loading…
Cancel
Save