diff --git "a/app.py" "b/app.py" deleted file mode 100644--- "a/app.py" +++ /dev/null @@ -1,2330 +0,0 @@ - -import os -import io -import base64 -import json -import logging -import threading -import time -from datetime import datetime -from uuid import uuid4 - -from flask import Flask, render_template_string, request, redirect, url_for, flash, jsonify, Response -from PIL import Image -import google.generativeai as genai -import numpy as np -from huggingface_hub import HfApi, hf_hub_download -from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError -from werkzeug.utils import secure_filename -from dotenv import load_dotenv -import requests - -load_dotenv() - -app = Flask(__name__) -app.secret_key = 'your_unique_secret_key_oco_shop_12345_no_login' -DATA_FILE = 'data.json' - -SYNC_FILES = [DATA_FILE] - -REPO_ID = "Kgshop/oco" -HF_TOKEN_WRITE = os.getenv("HF_TOKEN") -HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") -GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") - -STORE_ADDRESS = "Рынок Кербент, 6 ряд , 43 контейнер " -WHATSAPP_NUMBER = "+996779075126" - -CURRENCY_CODE = 'KGS' -CURRENCY_NAME = 'Кыргызский сом' - -DOWNLOAD_RETRIES = 3 -DOWNLOAD_DELAY = 5 - -logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') - -def download_db_from_hf(specific_file=None, retries=DOWNLOAD_RETRIES, delay=DOWNLOAD_DELAY): - if not HF_TOKEN_READ and not HF_TOKEN_WRITE: - logging.warning("HF_TOKEN_READ/HF_TOKEN_WRITE not set. Download might fail for private repos.") - - token_to_use = HF_TOKEN_READ if HF_TOKEN_READ else HF_TOKEN_WRITE - - files_to_download = [specific_file] if specific_file else SYNC_FILES - all_successful = True - - for file_name in files_to_download: - success = False - for attempt in range(retries + 1): - try: - local_path = hf_hub_download( - repo_id=REPO_ID, - filename=file_name, - repo_type="dataset", - token=token_to_use, - local_dir=".", - local_dir_use_symlinks=False, - force_download=True, - resume_download=False - ) - success = True - break - except RepositoryNotFoundError: - return False - except HfHubHTTPError as e: - if e.response.status_code == 404: - if attempt == 0 and not os.path.exists(file_name): - try: - if file_name == DATA_FILE: - with open(file_name, 'w', encoding='utf-8') as f: - json.dump({'products': [], 'categories': [], 'orders': {}, 'organization_info': {}}, f) - except Exception as create_e: - pass - success = False - break - else: - pass - except requests.exceptions.RequestException as e: - pass - except Exception as e: - pass - - if attempt < retries: - time.sleep(delay) - - if not success: - all_successful = False - - return all_successful - -def upload_db_to_hf(specific_file=None): - if not HF_TOKEN_WRITE: - return - try: - api = HfApi() - files_to_upload = [specific_file] if specific_file else SYNC_FILES - - for file_name in files_to_upload: - if os.path.exists(file_name): - try: - api.upload_file( - path_or_fileobj=file_name, - path_in_repo=file_name, - repo_id=REPO_ID, - repo_type="dataset", - token=HF_TOKEN_WRITE, - commit_message=f"Sync {file_name} {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" - ) - except Exception as e: - pass - else: - pass - except Exception as e: - pass - -def periodic_backup(): - backup_interval = 1800 - while True: - time.sleep(backup_interval) - upload_db_to_hf() - -def load_data(): - default_organization_info = { - "about_us": "Мы — O&CO, ваш надежный партнер в мире уникальных товаров. Мы предлагаем широкий ассортимент продукции, от электроники до товаров для дома, всегда стремясь к качеству и доступности. Наша миссия — сделать ваш шопинг приятным и удобным, предлагая только лучшие товары, тщательно отобранные для вас.", - "shipping": "Доставка осуществляется по всему Кыргызстану. Стоимость и сроки доставки зависят от региона и веса товара. По Бишкеку доставка возможна в течение 1-2 рабочих дней, в регионы — от 3 до 7 дней. Для уточнения деталей свяжитесь с нами.", - "returns": "Возврат и обмен товара возможен в течение 14 дней с момента покупки, при условии сохранения товарного вида, упаковки и чека. Некоторые категории товаров могут иметь особые условия возврата. Пожалуйста, свяжитесь с нами для оформления возврата или обмена.", - "contact": f"Наш магазин находится по адресу: {STORE_ADDRESS}. Связаться с нами можно по телефону: {WHATSAPP_NUMBER} или через WhatsApp по этому же номеру. Мы работаем ежедневно с 9:00 до 18:00." - } - default_data = {'products': [], 'categories': [], 'orders': {}, 'organization_info': default_organization_info} - data = default_data - try: - with open(DATA_FILE, 'r', encoding='utf-8') as file: - data = json.load(file) - if not isinstance(data, dict): - raise FileNotFoundError - if 'products' not in data: data['products'] = [] - if 'categories' not in data: data['categories'] = [] - if 'orders' not in data: data['orders'] = {} - if 'organization_info' not in data: data['organization_info'] = default_organization_info - except FileNotFoundError: - if download_db_from_hf(specific_file=DATA_FILE): - try: - with open(DATA_FILE, 'r', encoding='utf-8') as file: - data = json.load(file) - if not isinstance(data, dict): - data = default_data - if 'products' not in data: data['products'] = [] - if 'categories' not in data: data['categories'] = [] - if 'orders' not in data: data['orders'] = {} - if 'organization_info' not in data: data['organization_info'] = default_organization_info - except (FileNotFoundError, json.JSONDecodeError, Exception) as e: - data = default_data - else: - data = default_data - except json.JSONDecodeError: - if download_db_from_hf(specific_file=DATA_FILE): - try: - with open(DATA_FILE, 'r', encoding='utf-8') as file: - data = json.load(file) - if not isinstance(data, dict): - data = default_data - if 'products' not in data: data['products'] = [] - if 'categories' not in data: data['categories'] = [] - if 'orders' not in data: data['orders'] = {} - if 'organization_info' not in data: data['organization_info'] = default_organization_info - except (FileNotFoundError, json.JSONDecodeError, Exception) as e: - data = default_data - else: - data = default_data - except Exception as e: - data = default_data - - for product in data['products']: - if 'product_id' not in product: - product['product_id'] = uuid4().hex - if any('product_id' not in p for p in data['products']): - save_data(data) - - if not os.path.exists(DATA_FILE): - try: - with open(DATA_FILE, 'w', encoding='utf-8') as f: - json.dump(default_data, f) - except Exception as create_e: - pass - return data - -def save_data(data): - try: - if not isinstance(data, dict): - return - if 'products' not in data: data['products'] = [] - if 'categories' not in data: data['categories'] = [] - if 'orders' not in data: data['orders'] = {} - if 'organization_info' not in data: data['organization_info'] = {} - - with open(DATA_FILE, 'w', encoding='utf-8') as file: - json.dump(data, file, ensure_ascii=False, indent=4) - upload_db_to_hf(specific_file=DATA_FILE) - except Exception as e: - pass - -def configure_gemini(): - if not GOOGLE_API_KEY: - return False - try: - genai.configure(api_key=GOOGLE_API_KEY) - return True - except Exception as e: - return False - -def generate_ai_description_from_image(image_data, language): - if not configure_gemini(): - raise ValueError("Google AI API не настроен.") - - try: - if not image_data: - raise ValueError("Файл изображения не найден.") - image_stream = io.BytesIO(image_data) - image = Image.open(image_stream).convert('RGB') - except Exception as e: - raise ValueError(f"Не удалось обработать изображение. Убедитесь, что это действительный файл изображения.") - - base_prompt = "Напиши большой и красивый, содержательный рекламный пост минимум на 1000 символов со смайликами и 25 тематических хэштегов с ключевыми словами разных вариантов, чтобы мои клиенты могли найти меня в поиске Instagram, Google и т.д. по ключевым словам. Пост пиши исключительно под товар, который на фото, без адресов и номеров телефона." - - lang_suffix = "" - if language == "Русский": - lang_suffix = " Пиши на русском языке." - elif language == "Кыргызский": - lang_suffix = " Пиши на кыргызском языке." - elif language == "Казахский": - lang_suffix = " Пиши на казахском языке." - elif language == "Узбекский": - lang_suffix = " Пиши на узбекском языке." - - final_prompt = f"{base_prompt}{lang_suffix}" - - try: - model = genai.GenerativeModel('gemma-3-27b-it') - - response = model.generate_content([final_prompt, image]) - - if hasattr(response, 'text'): - return response.text - else: - if response.parts: - return "".join(part.text for part in response.parts if hasattr(part, 'text')) - else: - response.resolve() - return response.text - - except Exception as e: - if "API key not valid" in str(e): - raise ValueError("Внутренняя ошибка конфигурации API.") - elif " Billing account not found" in str(e): - raise ValueError("Проблема с биллингом аккаунта Google Cloud. Проверьте ваш аккаунт.") - elif "Could not find model" in str(e): - raise ValueError(f"Модель 'learnlm-2.0-flash-experimental' не найдена или недоступна.") - elif "resource has been exhausted" in str(e).lower(): - raise ValueError("Квота запросов исчерпана. Попробуйте позже.") - elif "content has been blocked" in str(e).lower(): - reason = "неизвестна" - if hasattr(e, 'response') and hasattr(e.response, 'prompt_feedback') and e.response.prompt_feedback.block_reason: - reason = e.response.prompt_feedback.block_reason - raise ValueError(f"Генерация контента заблокирована из-за политики безопасности (причина: {reason}). Попробуйте другое изображение или запрос.") - else: - raise ValueError(f"Ошибка при генерации контента: {e}") - -def generate_chat_response(message, chat_history_from_client): - if not configure_gemini(): - return "Извините, сервис чата временно недоступен. Пожалуйста, попробуйте позже." - - data = load_data() - products = data.get('products', []) - categories = data.get('categories', []) - organization_info = data.get('organization_info', {}) - - product_info_list = [] - for p in products: - if p.get('in_stock', True): - price_display = f"{p.get('price', 0):.2f}".replace('.00', '') - # AI is instructed to output in format: [ID_ТОВАРА: Название: ] - # We add a placeholder here for the AI to pick up on it. - product_info_list.append(f"- [ID_ТОВАРА: {p.get('product_id', 'N/A')} Название: {p.get('name', 'Без названия')}], Категория: {p.get('category', 'Без категории')}, Цена: {price_display} {CURRENCY_CODE}, Описание: {p.get('description', '')[:100]}...") - product_list_str = "\n".join(product_info_list) if product_info_list else "В данный момент нет товаров в наличии." - - category_list_str = ", ".join(categories) if categories else "Категорий пока нет." - - org_info_str = "" - if organization_info: - org_info_str += "\n\nИнформация о магазине:\n" - if organization_info.get("about_us"): - org_info_str += f"О нас: {organization_info['about_us']}\n" - if organization_info.get("shipping"): - org_info_str += f"Доставка: {organization_info['shipping']}\n" - if organization_info.get("returns"): - org_info_str += f"Возврат и обмен: {organization_info['returns']}\n" - if organization_info.get("contact"): - org_info_str += f"Контактная информация: {organization_info['contact']}\n" - - - system_instruction_content = ( - "Ты - доброжелательный и очень полезный виртуальный консультант для магазина O&CO. " - "Твоя задача - помогать пользователям находить товары, отвечать на вопросы о них, предлагать варианты, а также предоставлять информацию о магазине. " - "Всегда будь вежлив, информативен и стремись решить проблему пользователя. " - "Никогда не выдумывай товары или категории, которых нет в предоставленных списках. " - "Когда ты предлагаешь товар, всегда указывай его название и ID, используя *точный формат*: [ID_ТОВАРА: Название: ]. Это *очень важно* для клиента. " - "Если пользователь ищет товар или категорию, предлагай несколько наиболее подходящих вариантов или перечисляй доступные из этой категории.\n\n" - f"Список доступных категорий: {category_list_str}.\n\n" - f"Список доступных товаров в магазине:\n" - f"{product_list_str}" - f"{org_info_str}\n\n" - "Если пользователь спрашивает про товары или категории, которых нет в списках, вежливо сообщи, что таких товаров/категорий нет и предложи что-то из имеющихся, или перечисли доступные категории. " - "Если вопрос касается общей информации о магазине (например, 'о нас', 'доставка', 'возврат', 'контакты'), используй данные из блока 'Информация о магазине'. " - "Старайся быть кратким, но информативным. Используй эмодзи для дружелюбности. " - "Избегай упоминания Hugging Face или Hugging Face Hub." - ) - - generated_text = "" - response = None - - try: - model = genai.GenerativeModel('gemma-3-27b-it', system_instruction=system_instruction_content) - - model_chat_history_for_gemini = [] - for entry in chat_history_from_client: - gemini_role = 'model' if entry['role'] == 'ai' else 'user' - model_chat_history_for_gemini.append({ - 'role': gemini_role, - 'parts': [{'text': entry['text']}] - }) - - chat = model.start_chat(history=model_chat_history_for_gemini) - - response = chat.send_message(message, generation_config={'max_output_tokens': 1000}) - - if hasattr(response, 'text'): - generated_text = response.text - elif response.parts: - generated_text = "".join(part.text for part in response.parts if hasattr(part, 'text')) - else: - response.resolve() - if hasattr(response, 'text'): - generated_text = response.text - else: - raise ValueError("AI did not return a valid text response.") - - return generated_text - - except Exception as e: - if "API key not valid" in str(e): - return "Внутренняя ошибка конфигурации API." - elif " Billing account not found" in str(e): - return "Проблема с биллингом аккаунта Google Cloud." - elif "Could not find model" in str(e): - return "Модель 'learnlm-2.0-flash-experimental' не найдена или недоступна." - elif "resource has been exhausted" in str(e).lower(): - return "Квота запросов исчерпана. Попробуйте позже." - elif "content has been blocked" in str(e).lower() or (response is not None and hasattr(response, 'prompt_feedback') and response.prompt_feedback.block_reason): - reason = response.prompt_feedback.block_reason if (response and hasattr(response, 'prompt_feedback')) else "неизвестна" - return f"Извините, Ваш запрос был заблокирован из-за политики безопасности (причина: {reason}). Пожалуйста, переформулируйте его." - else: - return f"Извините, произошла ошибка: {e}" - -CATALOG_TEMPLATE = ''' - - - - - - O&CO - Каталог - - - - - - -
-
-
- - -
- - - -
- -
- {% set has_products = False %} - {% for category_name in ordered_categories %} - {% if products_by_category[category_name] %} - {% set has_products = True %} -
-
-

{{ category_name }}

- > -
-
- {% for product in products_by_category[category_name] %} -
- {% if product.get('is_top', False) %} - - {% endif %} -
- {% if product.get('photos') and product['photos']|length > 0 %} - {{ product.name }} - {% else %} - No Image - {% endif %} -
-
- {{ "%.2f"|format(product.price) }} {{ currency_code }} -
-
- {% endfor %} -
-
- {% endif %} - {% endfor %} - - {% if not has_products %} -

Товары пока не добавлены.

- {% endif %} - -
-
- - - - - - - - - -
- - -
- -
- - - - - -''' - -PRODUCT_DETAIL_TEMPLATE = ''' -
-

{{ product['name'] }}

-
-
- {% if product.get('photos') and product['photos']|length > 0 %} - {% for photo in product['photos'] %} -
-
- {{ product['name'] }} - фото {{ loop.index }} -
-
- {% endfor %} - {% else %} -
- Изображение отсутствует -
- {% endif %} -
- {% if product.get('photos') and product['photos']|length > 1 %} -
-
-
- {% endif %} -
- -
-

Цена: {{ "%.2f"|format(product.price) }} {{ currency_code }}

- -
- -
-

Категория: {{ product.get('category', 'Без категории') }}

-

Описание:
{{ product.get('description', 'Описание отсутствует.')|replace('\\n', '
')|safe }}

- {% set colors = product.get('colors', []) %} - {% if colors and colors|select('ne', '')|list|length > 0 %} -

Доступные цвета/варианты: {{ colors|select('ne', '')|join(', ') }}

- {% endif %} -
-
-''' - -ORDER_TEMPLATE = ''' - - - - - - Заказ №{{ order.id }} - O&CO - - - - - -
- {% if order %} -

Ваш Заказ №{{ order.id }}

-

Дата создания: {{ order.created_at }}

- -

Товары в заказе

-
- {% for item in order.cart %} -
- {{ item.name }} -
- {{ item.name }} {% if item.color != 'N/A' %}({{ item.color }}){% endif %} - {{ "%.2f"|format(item.price) }} {{ currency_code }} × {{ item.quantity }} -
-
- {{ "%.2f"|format(item.price * item.quantity) }} {{ currency_code }} -
-
- {% endfor %} -
- -
-

Общая сумма товаров: {{ "%.2f"|format(order.total_price) }} {{ currency_code }}

-

ИТОГО К ОПЛАТЕ: {{ "%.2f"|format(order.total_price) }} {{ currency_code }}

-
- -
-

Статус заказа

-

Этот заказ был оформлен без входа в систему.

-

Пожалуйста, свяжитесь с нами по WhatsApp для подтверждения и уточнения деталей.

-
- -
- -
- - ← Вернуться в каталог - - - - {% else %} -

Ошибка

-

Заказ с таким ID не найден.

- ← Вернуться в каталог - {% endif %} -
- - -''' - -ADMIN_TEMPLATE = ''' - - - - - - Админ-панель - O&CO - - - - - -
-
-
- O&CO Logo -

Админ-панель O&CO

-
- Перейти в каталог -
- - {% with messages = get_flashed_messages(with_categories=true) %} - {% if messages %} - {% for category, message in messages %} -
{{ message }}
- {% endfor %} - {% endif %} - {% endwith %} - -
-

Синхронизация с Датацентром

-
-
- -
-
- -
-
-

Резервное копирование происходит автоматически каждые 30 минут, а также после каждого сохранения данных. Используйте эти кнопки для немедленной синхронизации.

-
- -
-
-
-

Управление категориями

-
- Добавить новую категорию -
-
- - - - -
-
-
- -

Существующие категории:

- {% if categories %} -
- {% for category in categories %} -
- {{ category }} -
- - - -
-
- {% endfor %} -
- {% else %} -

Категорий пока нет.

- {% endif %} -
-
- -
-
-

Информация о магазине

-
- Развернуть/Свернуть -
-
- - - - - - - - - - -
-

Эта информация будет использоваться ИИ-ассистентом для ответов на вопросы о вашем магазине.

-
-
-
-
-
- -
-

Управление товарами

-
- Добавить новый товар -
-
- - - - - - - - - - - - - - - -
-
- - -
-
- -
-
- - -
-
- - -
-
- -
-
-
- -

Список товаров:

- {% if products %} -
- {% for product in products %} -
-
-
- {% if product.get('photos') %} - - Фото - - {% else %} - Нет фото - {% endif %} -
-
-

- {{ product['name'] }} - {% if product.get('in_stock', True) %} - В наличии - {% else %} - Нет в наличии - {% endif %} - {% if product.get('is_top', False) %} - Топ - {% endif %} -

-

Категория: {{ product.get('category', 'Без категории') }}

-

Цена: {{ "%.2f"|format(product.price) }} {{ currency_code }}

-

Описание: {{ product.get('description', 'N/A')[:150] }}{% if product.get('description', '')|length > 150 %}...{% endif %}

- {% set colors = product.get('colors', []) %} -

Цвета/Вар-ts: {{ colors|select('ne', '')|join(', ') if colors|select('ne', '')|list|length > 0 else 'Нет' }}

- {% if product.get('photos') and product['photos']|length > 1 %} -

(Всего фото: {{ product['photos']|length }})

- {% endif %} -
-
- -
- -
- - - -
-
- -
-

Редактирование: {{ product['name'] }}

-
- - - - - - - - - {% if product.get('photos') %} -

Текущие фото:

-
- {% for photo in product['photos'] %} - Фото {{ loop.index }} - {% endfor %} -
- {% endif %} - - - - - - - - -
- {% set current_colors = product.get('colors', []) %} - {% if current_colors and current_colors|select('ne', '')|list|length > 0 %} - {% for color in current_colors %} - {% if color.strip() %} -
- - -
- {% endif %} - {% endfor %} - {% else %} -
- - -
- {% endif %} -
- -
-
- - -
-
- - -
-
- -
-
-
- {% endfor %} -
- {% else %} -

Товаров пока нет.

- {% endif %} -
- -
- - - - -''' - -@app.route('/') -def catalog(): - data = load_data() - all_products_raw = data.get('products', []) - - product_categories = set(p.get('category', 'Без категории') for p in all_products_raw) - admin_categories = set(data.get('categories', [])) - all_cat_names = sorted(list(product_categories.union(admin_categories))) - - products_in_stock = [p for p in all_products_raw if p.get('in_stock', True)] - - # Sort products for consistent display and indexing - products_sorted_for_js = sorted(products_in_stock, key=lambda p: (not p.get('is_top', False), p.get('name', '').lower())) - - products_by_category = {cat: [] for cat in all_cat_names} - for product in products_in_stock: - products_by_category[product.get('category', 'Без категории')].append(product) - - for category in products_by_category: - products_by_category[category].sort(key=lambda p: (not p.get('is_top', False), p.get('name', '').lower())) - - ordered_categories = [cat for cat in all_cat_names if products_by_category.get(cat)] - - return render_template_string( - CATALOG_TEMPLATE, - products_by_category=products_by_category, - ordered_categories=ordered_categories, - products_json=json.dumps(products_sorted_for_js), # Pass sorted products to JS - repo_id=REPO_ID, - store_address=STORE_ADDRESS, - currency_code=CURRENCY_CODE - ) - - -@app.route('/product/') -def product_detail(index): - data = load_data() - all_products_raw = data.get('products', []) - products_in_stock = [p for p in all_products_raw if p.get('in_stock', True)] - products_sorted = sorted(products_in_stock, key=lambda p: (not p.get('is_top', False), p.get('name', '').lower())) - - try: - product = products_sorted[index] - except IndexError: - return "Товар не найден или отсутствует в наличии.", 404 - - return render_template_string( - PRODUCT_DETAIL_TEMPLATE, - product=product, - repo_id=REPO_ID, - currency_code=CURRENCY_CODE - ) - -@app.route('/create_order', methods=['POST']) -def create_order(): - order_data = request.get_json() - - if not order_data or 'cart' not in order_data or not order_data['cart']: - return jsonify({"error": "Корзина пуста или не передана."}), 400 - - cart_items = order_data['cart'] - - total_price = 0 - processed_cart = [] - for item in cart_items: - if not all(k in item for k in ('name', 'price', 'quantity')): - return jsonify({"error": "Неверный формат товара в корзине."}), 400 - try: - price = float(item['price']) - quantity = int(item['quantity']) - if price < 0 or quantity <= 0: - raise ValueError("Invalid price or quantity") - processed_cart.append({ - "product_id": item.get('product_id', 'N/A'), - "name": item['name'], - "price": price, - "quantity": quantity, - "color": item.get('color', 'N/A'), - "photo": item.get('photo'), - "photo_url": f"https://huggingface.co/datasets/{REPO_ID}/resolve/main/photos/{item['photo']}" if item.get('photo') else "https://via.placeholder.com/60x60.png?text=N/A" - }) - total_price += price * quantity - except (ValueError, TypeError) as e: - return jsonify({"error": "Неверная цена или количество в то��аре."}), 400 - - order_id = f"{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid4().hex[:6]}" - order_timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') - - new_order = { - "id": order_id, - "created_at": order_timestamp, - "cart": processed_cart, - "total_price": round(total_price, 2), - "user_info": None, - "status": "new" - } - - try: - data = load_data() - if 'orders' not in data or not isinstance(data.get('orders'), dict): - data['orders'] = {} - - data['orders'][order_id] = new_order - save_data(data) - return jsonify({"order_id": order_id}), 201 - - except Exception as e: - return jsonify({"error": "Ошибка сервера при сохранении заказа."}), 500 - - -@app.route('/order/') -def view_order(order_id): - data = load_data() - order = data.get('orders', {}).get(order_id) - - return render_template_string(ORDER_TEMPLATE, - order=order, - repo_id=REPO_ID, - currency_code=CURRENCY_CODE, - whatsapp_number=WHATSAPP_NUMBER) - - -@app.route('/admin', methods=['GET', 'POST']) -def admin(): - data = load_data() - products = data.get('products', []) - categories = data.get('categories', []) - organization_info = data.get('organization_info', {}) - - if 'orders' not in data or not isinstance(data.get('orders'), dict): - data['orders'] = {} - - if request.method == 'POST': - action = request.form.get('action') - - try: - if action == 'add_category': - category_name = request.form.get('category_name', '').strip() - if category_name and category_name not in categories: - categories.append(category_name) - data['categories'] = categories - save_data(data) - flash(f"Категория '{category_name}' успешно добавлена.", 'success') - elif not category_name: - flash("Название категории не может быть пустым.", 'error') - else: - flash(f"Категория '{category_name}' уже существует.", 'error') - - elif action == 'delete_category': - category_to_delete = request.form.get('category_name') - if category_to_delete and category_to_delete in categories: - categories.remove(category_to_delete) - updated_count = 0 - for product in products: - if product.get('category') == category_to_delete: - product['category'] = 'Без категории' - updated_count += 1 - data['categories'] = categories - data['products'] = products - save_data(data) - flash(f"Категория '{category_to_delete}' удалена. {updated_count} товаров обновлено.", 'success') - else: - flash(f"Не удалось удалить категорию '{category_to_delete}'.", 'error') - - elif action == 'update_org_info': - organization_info['about_us'] = request.form.get('about_us', '').strip() - organization_info['shipping'] = request.form.get('shipping', '').strip() - organization_info['returns'] = request.form.get('returns', '').strip() - organization_info['contact'] = request.form.get('contact', '').strip() - data['organization_info'] = organization_info - save_data(data) - flash("Информация о магазине успешно обновлена.", 'success') - - elif action == 'add_product': - name = request.form.get('name', '').strip() - price_str = request.form.get('price', '').replace(',', '.') - description = request.form.get('description', '').strip() - category = request.form.get('category') - photos_files = request.files.getlist('photos') - colors = [c.strip() for c in request.form.getlist('colors') if c.strip()] - in_stock = 'in_stock' in request.form - is_top = 'is_top' in request.form - - if not name or not price_str: - flash("Название и цена товара обязательны.", 'error') - return redirect(url_for('admin')) - - try: - price = round(float(price_str), 2) - if price < 0: price = 0 - except ValueError: - flash("Неверный формат цены.", 'error') - return redirect(url_for('admin')) - - photos_list = [] - if photos_files and HF_TOKEN_WRITE: - uploads_dir = 'uploads_temp' - os.makedirs(uploads_dir, exist_ok=True) - api = HfApi() - photo_limit = 10 - uploaded_count = 0 - for photo in photos_files: - if uploaded_count >= photo_limit: - flash(f"Загружено только первые {photo_limit} фото.", "warning") - break - if photo and photo.filename: - try: - ext = os.path.splitext(photo.filename)[1].lower() - if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: - flash(f"Файл {photo.filename} не является изображением и был пропущен.", "warning") - continue - - safe_name = secure_filename(name.replace(' ', '_'))[:50] - photo_filename = f"{safe_name}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}{ext}" - temp_path = os.path.join(uploads_dir, photo_filename) - photo.save(temp_path) - api.upload_file( - path_or_fileobj=temp_path, - path_in_repo=f"photos/{photo_filename}", - repo_id=REPO_ID, - repo_type="dataset", - token=HF_TOKEN_WRITE, - commit_message=f"Add photo for product {name}" - ) - photos_list.append(photo_filename) - os.remove(temp_path) - uploaded_count += 1 - except Exception as e: - flash(f"Ошибка при загрузке фото {photo.filename}.", 'error') - if os.path.exists(temp_path): - try: os.remove(temp_path) - except OSError: pass - elif photo and not photo.filename: - pass - try: - if os.path.exists(uploads_dir) and not os.listdir(uploads_dir): - os.rmdir(uploads_dir) - except OSError as e: - pass - elif not HF_TOKEN_WRITE and photos_files and any(f.filename for f in photos_files): - flash("HF_TOKEN (write) не настроен. Фотографии не были загружены.", "warning") - - - new_product = { - 'product_id': uuid4().hex, - 'name': name, 'price': price, 'description': description, - 'category': category if category in categories else 'Без категории', - 'photos': photos_list, 'colors': colors, - 'in_stock': in_stock, 'is_top': is_top - } - products.append(new_product) - data['products'] = products - save_data(data) - flash(f"Товар '{name}' успешно добавлен.", 'success') - - elif action == 'edit_product': - product_id = request.form.get('product_id') - product_to_edit = next((p for p in products if p.get('product_id') == product_id), None) - - if product_to_edit is None: - flash(f"Ошибка редактирования: товар с ID '{product_id}' не найден.", 'error') - return redirect(url_for('admin')) - - original_name = product_to_edit.get('name', 'N/A') - - product_to_edit['name'] = request.form.get('name', product_to_edit['name']).strip() - price_str = request.form.get('price', str(product_to_edit['price'])).replace(',', '.') - product_to_edit['description'] = request.form.get('description', product_to_edit.get('description', '')).strip() - category = request.form.get('category') - product_to_edit['category'] = category if category in categories else 'Без категории' - product_to_edit['colors'] = [c.strip() for c in request.form.getlist('colors') if c.strip()] - product_to_edit['in_stock'] = 'in_stock' in request.form - product_to_edit['is_top'] = 'is_top' in request.form - - try: - price = round(float(price_str), 2) - if price < 0: price = 0 - product_to_edit['price'] = price - except ValueError: - flash(f"Неверный формат це��ы для товара '{original_name}'. Цена не изменена.", 'warning') - - photos_files = request.files.getlist('photos') - if photos_files and any(f.filename for f in photos_files) and HF_TOKEN_WRITE: - uploads_dir = 'uploads_temp' - os.makedirs(uploads_dir, exist_ok=True) - api = HfApi() - new_photos_list = [] - photo_limit = 10 - uploaded_count = 0 - for photo in photos_files: - if uploaded_count >= photo_limit: - flash(f"Загружено только первые {photo_limit} фото.", "warning") - break - if photo and photo.filename: - try: - ext = os.path.splitext(photo.filename)[1].lower() - if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: - flash(f"Файл {photo.filename} не является изображением и был пропущен.", "warning") - continue - - safe_name = secure_filename(product_to_edit['name'].replace(' ', '_'))[:50] - photo_filename = f"{safe_name}_{datetime.now().strftime('%Y%m%d%H%M%S%f')}{ext}" - temp_path = os.path.join(uploads_dir, photo_filename) - photo.save(temp_path) - api.upload_file(path_or_fileobj=temp_path, path_in_repo=f"photos/{photo_filename}", - repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, - commit_message=f"Update photo for product {product_to_edit['name']}") - new_photos_list.append(photo_filename) - os.remove(temp_path) - uploaded_count += 1 - except Exception as e: - flash(f"Ошибка при загрузке нового фото {photo.filename}.", 'error') - if os.path.exists(temp_path): - try: os.remove(temp_path) - except OSError: pass - try: - if os.path.exists(uploads_dir) and not os.listdir(uploads_dir): - os.rmdir(uploads_dir) - except OSError as e: - pass - - if new_photos_list: - old_photos = product_to_edit.get('photos', []) - if old_photos: - try: - api = HfApi() - api.delete_files( - repo_id=REPO_ID, - paths_in_repo=[f"photos/{p}" for p in old_photos], - repo_type="dataset", - token=HF_TOKEN_WRITE, - commit_message=f"Delete old photos for product {product_to_edit['name']}" - ) - except Exception as e: - flash("Не удалось удалить старые фотографии с сервера. Новые фото загружены.", "warning") - product_to_edit['photos'] = new_photos_list - flash("Фотографии товара успешно обновлены.", "success") - elif uploaded_count == 0 and any(f.filename for f in photos_files): - flash("Не удалось загрузить новые фотографии (возможно, неверный формат).", "error") - elif not HF_TOKEN_WRITE and photos_files and any(f.filename for f in photos_files): - flash("HF_TOKEN (write) не настроен. Фотографии не были обновлены.", "warning") - - save_data(data) - flash(f"Товар '{product_to_edit['name']}' успешно обновлен.", 'success') - - elif action == 'delete_product': - product_id = request.form.get('product_id') - product_index = next((i for i, p in enumerate(products) if p.get('product_id') == product_id), -1) - - if product_index == -1: - flash(f"Ошибка удаления: товар с ID '{product_id}' не найден.", 'error') - return redirect(url_for('admin')) - - deleted_product = products.pop(product_index) - product_name = deleted_product.get('name', 'N/A') - - photos_to_delete = deleted_product.get('photos', []) - if photos_to_delete and HF_TOKEN_WRITE: - try: - api = HfApi() - api.delete_files( - repo_id=REPO_ID, - paths_in_repo=[f"photos/{p}" for p in photos_to_delete], - repo_type="dataset", - token=HF_TOKEN_WRITE, - commit_message=f"Delete photos for deleted product {product_name}" - ) - except Exception as e: - flash(f"Не удалось удалить фото для товара '{product_name}' с сервера. Товар удален локально.", "warning") - elif photos_to_delete and not HF_TOKEN_WRITE: - flash(f"Товар '{product_name}' удален локально, но фото не удалены с сервера (токен не задан).", "warning") - - data['products'] = products - save_data(data) - flash(f"Товар '{product_name}' удален.", 'success') - - else: - flash(f"Неизвестное действие: {action}", 'warning') - - return redirect(url_for('admin')) - - except Exception as e: - flash(f"Произошла внутренняя ошибка при выполнении действия '{action}'. Подробности в логе сервера.", 'error') - return redirect(url_for('admin')) - - current_data = load_data() - display_products = sorted(current_data.get('products', []), key=lambda p: p.get('name', '').lower()) - display_categories = sorted(current_data.get('categories', [])) - display_organization_info = current_data.get('organization_info', {}) - - return render_template_string( - ADMIN_TEMPLATE, - products=display_products, - categories=display_categories, - organization_info=display_organization_info, - repo_id=REPO_ID, - currency_code=CURRENCY_CODE - ) - -@app.route('/generate_description_ai', methods=['POST']) -def handle_generate_description_ai(): - request_data = request.get_json() - base64_image = request_data.get('image') - language = request_data.get('language', 'Русский') - - if not base64_image: - return jsonify({"error": "Изображение не найдено в запросе."}), 400 - - try: - image_data = base64.b64decode(base64_image) - result_text = generate_ai_description_from_image(image_data, language) - return jsonify({"text": result_text}) - except ValueError as ve: - return jsonify({"error": str(ve)}), 400 - except Exception as e: - return jsonify({"error": f"Внутренняя ошибка сервера: {e}"}), 500 - -@app.route('/chat_with_ai', methods=['POST']) -def handle_chat_with_ai(): - request_data = request.get_json() - user_message = request_data.get('message') - chat_history_from_client = request_data.get('history', []) - - if not user_message: - return jsonify({"error": "Сообщение не может быть пустым."}), 400 - - try: - ai_response_text = generate_chat_response(user_message, chat_history_from_client) - return jsonify({"text": ai_response_text}) - except Exception as e: - return jsonify({"error": f"Ошибка чата: {e}"}), 500 - -@app.route('/force_upload', methods=['POST']) -def force_upload(): - try: - upload_db_to_hf() - flash("Данные успешно загружены на Hugging Face.", 'success') - except Exception as e: - flash(f"Ошибка при загрузке на Hugging Face: {e}", 'error') - return redirect(url_for('admin')) - -@app.route('/force_download', methods=['POST']) -def force_download(): - try: - if download_db_from_hf(): - flash("Данные успешно скачаны с Hugging Face. Локальные файлы обновлены.", 'success') - load_data() - else: - flash("Не удалось скачать данные с Hugging Face после нескольких попыток. Проверьте логи.", 'error') - except Exception as e: - flash(f"Ошибка при скачивании с Hugging Face: {e}", 'error') - return redirect(url_for('admin')) - -if __name__ == '__main__': - configure_gemini() - download_db_from_hf() - load_data() - - if HF_TOKEN_WRITE: - backup_thread = threading.Thread(target=periodic_backup, daemon=True) - backup_thread.start() - else: - pass - - port = int(os.environ.get('PORT', 7860)) - app.run(debug=False, host='0.0.0.0', port=port)