from flask import Flask, render_template_string, request, redirect, url_for, session, send_from_directory import json import os import logging import threading import time from datetime import datetime from huggingface_hub import HfApi, hf_hub_download from huggingface_hub.utils import RepositoryNotFoundError from werkzeug.utils import secure_filename import shutil # Import shutil for file operations app = Flask(__name__) app.secret_key = 'your_unique_secret_key_soola_cosmetics_7890' # Новый уникальный секретный ключ DATA_FILE = 'data_soola.json' USERS_FILE = 'users_soola.json' # CONFIG_FILE убран, так как курс фиксирован на KGS # Список файлов для синхронизации (убрали CONFIG_FILE) SYNC_FILES = [DATA_FILE, USERS_FILE] # Настройки Hugging Face REPO_ID = "Kgshop/Soola" # Оставляем старый? Или создать новый репозиторий? Уточнить. HF_TOKEN_WRITE = os.getenv("HF_TOKEN") HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") # Может быть тот же, что и WRITE # Адрес магазина (только один) STORE_ADDRESS = "Рынок Дордой, Джунхай, терминал, 38" # Поддерживаемые валюты - теперь только KGS CURRENCY_CODE = 'KGS' CURRENCY_NAME = 'Кыргызский сом (с)' # Настройка логирования logging.basicConfig(level=logging.INFO) # Можно изменить на DEBUG для подробного лога # --- Функции работы с данными и Hugging Face (без изменений в логике, кроме списка файлов) --- def load_data(): """Загружает данные товаров и категорий.""" try: # Попробуем скачать актуальные файлы перед чтением download_db_from_hf() except RepositoryNotFoundError: logging.warning(f"Репозиторий {REPO_ID} не найден на Hugging Face. Используется локальная версия, если есть.") except Exception as e: logging.warning(f"Ошибка при скачивании файлов с Hugging Face: {e}. Используется локальная версия, если есть.") try: with open(DATA_FILE, 'r', encoding='utf-8') as file: data = json.load(file) logging.info(f"Данные успешно загружены из {DATA_FILE}") # Проверка базовой структуры if not isinstance(data, dict): logging.warning(f"{DATA_FILE} не является словарем. Сброс к формату по умолчанию.") return {'products': [], 'categories': []} if 'products' not in data: data['products'] = [] logging.warning(f"Ключ 'products' отсутствовал в {DATA_FILE}. Добавлен пустой список.") if 'categories' not in data: data['categories'] = [] logging.warning(f"Ключ 'categories' отсутствовал в {DATA_FILE}. Добавлен пустой список.") return data except FileNotFoundError: logging.warning(f"Локальный файл {DATA_FILE} не найден. Создание структуры по умолчанию.") return {'products': [], 'categories': []} except json.JSONDecodeError: logging.error(f"Ошибка: Невозможно декодировать JSON из {DATA_FILE}. Возврат структуры по умолчанию.") # Попытка создать резервную копию поврежденного файла try: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") shutil.copyfile(DATA_FILE, f"{DATA_FILE}.corrupted_{timestamp}") logging.info(f"Создана резервная копия поврежденного файла: {DATA_FILE}.corrupted_{timestamp}") except Exception as copy_e: logging.error(f"Не удалось создать резервную копию поврежденного файла {DATA_FILE}: {copy_e}") return {'products': [], 'categories': []} except Exception as e: logging.error(f"Непредвиденная ошибка при загрузке данных из {DATA_FILE}: {e}") return {'products': [], 'categories': []} def save_data(data): """Сохраняет данные товаров и категорий.""" try: with open(DATA_FILE, 'w', encoding='utf-8') as file: json.dump(data, file, ensure_ascii=False, indent=4) logging.info(f"Данные успешно сохранены в {DATA_FILE}") # Загружаем на HF после локального сохранения upload_db_to_hf() except Exception as e: logging.error(f"Ошибка при сохранении данных в {DATA_FILE}: {e}") # Не пробрасываем исключение дальше, чтобы приложение могло продолжить работу, # но логируем ошибку. Важно отслеживать такие ошибки. def load_users(): """Загружает данные пользователей.""" try: # Попробуем скачать актуальные файлы перед чтением download_db_from_hf() # Убедимся, что работаем с последней версией except RepositoryNotFoundError: logging.warning(f"Репозиторий {REPO_ID} не найден на Hugging Face. Используется локальная версия users_soola.json, если есть.") except Exception as e: logging.warning(f"Ошибка при скачивании users_soola.json с Hugging Face: {e}. Используется локальная версия, если есть.") try: with open(USERS_FILE, 'r', encoding='utf-8') as file: users = json.load(file) logging.info(f"Данные пользователей успешно загружены из {USERS_FILE}") if not isinstance(users, dict): logging.warning(f"{USERS_FILE} не является словарем. Возвращается пустой словарь.") return {} return users except FileNotFoundError: logging.warning(f"Локальный файл {USERS_FILE} не найден. Возвращается пустой словарь.") return {} except json.JSONDecodeError: logging.error(f"Ошибка: Невозможно декодировать JSON из {USERS_FILE}. Возвращается пустой словарь.") # Попытка создать резервную копию поврежденного файла try: timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") shutil.copyfile(USERS_FILE, f"{USERS_FILE}.corrupted_{timestamp}") logging.info(f"Создана резервная копия поврежденного файла: {USERS_FILE}.corrupted_{timestamp}") except Exception as copy_e: logging.error(f"Не удалось создать резервную копию поврежденного файла {USERS_FILE}: {copy_e}") return {} except Exception as e: logging.error(f"Непредвиденная ошибка при загрузке данных из {USERS_FILE}: {e}") return {} def save_users(users): """Сохраняет данные пользователей.""" try: with open(USERS_FILE, 'w', encoding='utf-8') as file: json.dump(users, file, ensure_ascii=False, indent=4) logging.info(f"Данные пользователей успешно сохранены в {USERS_FILE}") # Загружаем на HF после локального сохранения upload_db_to_hf() except Exception as e: logging.error(f"Ошибка при сохранении данных пользователей в {USERS_FILE}: {e}") def upload_db_to_hf(): """Загружает файлы данных на Hugging Face.""" if not HF_TOKEN_WRITE: logging.warning("HF_TOKEN (токен для записи) не установлен. Загрузка на Hugging Face отключена.") return try: api = HfApi() logging.info(f"Попытка загрузки файлов {SYNC_FILES} в репозиторий {REPO_ID}...") for file_name in SYNC_FILES: if os.path.exists(file_name): 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"Автоматическое резервное копирование {file_name} {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ) logging.info(f"Файл {file_name} успешно загружен на Hugging Face.") else: logging.warning(f"Файл {file_name} не найден локально, пропуск загрузки.") except Exception as e: logging.error(f"Ошибка при загрузке файлов на Hugging Face: {e}") def download_db_from_hf(): """Скачивает файлы данных с Hugging Face.""" if not HF_TOKEN_READ: logging.warning("HF_TOKEN_READ (токен для чтения) не установлен. Скачивание с Hugging Face отключено.") # Не вызываем raise, чтобы приложение могло работать с локальными файлами, если они есть return try: api = HfApi() # Можно и без создания объекта api, hf_hub_download сам его использует logging.info(f"Попытка скачивания файлов {SYNC_FILES} из репозитория {REPO_ID}...") for file_name in SYNC_FILES: try: hf_hub_download( repo_id=REPO_ID, filename=file_name, repo_type="dataset", token=HF_TOKEN_READ, local_dir=".", # Скачиваем в текущую директорию local_dir_use_symlinks=False, # Важно для большинства сред развертывания force_download=True # Принудительно скачиваем, чтобы получить последнюю версию ) logging.info(f"Файл {file_name} успешно скачан из Hugging Face.") except RepositoryNotFoundError as repo_e: logging.error(f"Репозиторий {REPO_ID} не найден на Hugging Face: {repo_e}") raise # Пробрасываем ошибку, если репозиторий не найден except Exception as file_e: # Логируем ошибку для конкретного файла, но продолжаем скачивать остальные logging.error(f"Ошибка при скачивании файла {file_name}: {file_e}") except RepositoryNotFoundError: # Эта ошибка уже обработана выше, но на всякий случай ловим здесь тоже raise except Exception as e: # Общая ошибка при попытке скачивания logging.error(f"Общая ошибка при скачивании файлов с Hugging Face: {e}") # Не пробрасываем ошибку дальше, чтобы приложение могло попытаться запуститься с локальными данными def periodic_backup(): """Периодически загружает данные на Hugging Face.""" while True: logging.info("Запуск периодического резервного копирования...") upload_db_to_hf() logging.info("Периодическое резервное копирование завершено. Следующий запуск через 800 секунд.") time.sleep(800) # 13 минут 20 секунд # --- Маршруты Flask --- @app.route('/') def catalog(): """Главная страница каталога.""" data = load_data() products = data.get('products', []) categories = data.get('categories', []) is_authenticated = 'user' in session catalog_html = ''' Soola Cosmetics - Каталог

Soola Cosmetics

{% if is_authenticated %} Добро пожаловать, {{ session['user'] }}! Выйти {% else %} Войти {% endif %}
{{ store_address }}
{% for category in categories %} {% endfor %}
{% for product in products %}
{% if product.get('photos') and product['photos']|length > 0 %} {{ product.get('name', 'Фото товара') }} {% else %} {# Нет изображения #} {% endif %}

{{ product.get('name', 'Без названия') }}

{{ "%.2f"|format(product.get('price', 0)) }} {{ currency_code }}

{{ product.get('description', '') }}

{% if is_authenticated %} {% endif %}
{% endfor %}
''' return render_template_string( catalog_html, products=products, categories=categories, repo_id=REPO_ID, is_authenticated=is_authenticated, store_address=STORE_ADDRESS, session=session, # Передаем всю сессию currency_code=CURRENCY_CODE # Передаем код валюты # Убрали convert_price, selected_currency, currencies, kgs_to_usd ) @app.route('/product/') def product_detail(index): """Отображает детальную информацию о продукте в модальном окне.""" data = load_data() products = data.get('products', []) try: product = products[index] except IndexError: # Возвращаем HTML с сообщением об ошибке, чтобы он отобразился в модальном окне return "

Товар не найден.

", 404 # Используем Jinja для генерации HTML контента модального окна detail_html = '''
{# Обертка для содержимого #}

{{ product.get('name', 'Без названия') }}

{# Swiper контейнер для изображений #}
{% if product.get('photos') and product['photos']|length > 0 %} {% for photo in product['photos'] %}
{# Обертка для зума #} Фото {{ product.get('name', '') }} {# Ленивая загрузка #}
{% endfor %} {% else %} {# Слайд-заглушка, если фото нет #}
{# Нет изображения #}
{% endif %}
{# Элементы управления Swiper #} {% if product.get('photos') and product['photos']|length > 1 %} {# Показываем пагинацию/навигацию только если больше 1 фото #}
{% endif %}

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

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

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

{# Заменяем переносы строк на
#} {% if product.get('colors') and product['colors']|length > 0 %}

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

{% endif %}
''' # Передаем необходимые переменные в шаблон return render_template_string( detail_html, product=product, repo_id=REPO_ID, currency_code=CURRENCY_CODE # is_authenticated и convert_price убраны ) # Маршрут /set_currency убран # Маршрут /register убран @app.route('/login', methods=['GET', 'POST']) def login(): """Страница входа пользователя.""" if 'user' in session: return redirect(url_for('catalog')) # Если уже вошел, перенаправляем в каталог if request.method == 'POST': login = request.form.get('login') password = request.form.get('password') users = load_users() if login in users and users[login].get('password') == password: # Проверяем статус пользователя (если нужно будет добавить бан/деактивацию) if users[login].get('status', 'active') == 'active': session['user'] = login session['country'] = users[login].get('country', '') # Сохраняем доп. инфо в сессию session['city'] = users[login].get('city', '') # Валюта больше не устанавливается в сессию logging.info(f"Пользователь {login} успешно вошел в систему.") # Сохраняем логин в localStorage для возможного авто-входа # Делаем это через редирект с добавлением скрипта response = redirect(url_for('catalog')) # Установка cookie через response.set_cookie была бы надежнее, # но localStorage проще для примера авто-логина # Мы будем использовать JS на странице логина для сохранения return response else: logging.warning(f"Попытка входа заблокированного пользователя: {login}") error = "Ваш аккаунт неактивен. Обратитесь к администратору." return render_template_string(login_template, error=error) else: logging.warning(f"Неудачная попытка входа для логина: {login}") error = "Неверный логин или пароль." return render_template_string(login_template, error=error) # Передаем ошибку в шаблон # Отображаем форму входа (GET запрос) return render_template_string(login_template, error=None) # Передаем None как ошибку при первом заходе # Шаблон для страницы входа вынесен для читаемости login_template = ''' Вход - Soola Cosmetics

Вход в Soola Cosmetics

{% if error %}

{{ error }}

{% endif %}
Вернуться в каталог
''' @app.route('/auto_login', methods=['POST']) def auto_login(): """Обрабатывает попытку автоматического входа.""" if 'user' in session: return "Already logged in", 200 # Уже в системе data = request.get_json() if not data or 'login' not in data: return "Missing login", 400 login = data.get('login') users = load_users() if login in users: # Здесь не проверяем пароль, т.к. это авто-вход по сохраненному логину # Важно: Это менее безопасно! Используется для удобства. # Можно добавить проверку токена сессии из localStorage, если нужна безопасность. if users[login].get('status', 'active') == 'active': session['user'] = login session['country'] = users[login].get('country', '') session['city'] = users[login].get('city', '') # Валюта не устанавливается logging.info(f"Пользователь {login} автоматически вошел в систему.") return "OK", 200 else: logging.warning(f"Попытка авто-входа неактивного пользователя: {login}") return "User inactive", 403 # Forbidden else: logging.warning(f"Попытка авто-входа несуществующего пользователя: {login}") return "User not found", 404 @app.route('/logout') def logout(): """Выход пользователя из системы.""" logged_out_user = session.pop('user', None) session.pop('country', None) session.pop('city', None) # Валюта не удаляется, т.к. ее нет в сессии if logged_out_user: logging.info(f"Пользователь {logged_out_user} вышел из системы.") # Очищаем localStorage при выходе # Делаем это на клиенте после редиректа response = redirect(url_for('catalog')) # response.set_cookie('clearUser', 'true', max_age=5) # Альтернатива: использовать cookie # Проще добавить скрипт на страницу каталога, который проверит параметр? Или JS на этой странице return '''

Выход... Перенаправление в каталог.

'''.format(url_for('catalog'), url_for('catalog')) else: return redirect(url_for('catalog')) # Если не был залогинен, просто в каталог # --- Админ-панель --- @app.route('/admin', methods=['GET', 'POST']) def admin(): """Административная панель.""" # Простая проверка "админа" - первый зарегистрированный пользователь или заданный логин # В реальном приложении нужна система ролей! users = load_users() is_admin = False if 'user' in session: # Пример: Админ - это пользователь с логином 'admin' # Или можно сделать первого пользователя админом: # if users and session['user'] == next(iter(users)): # is_admin = True if session['user'] == 'admin': # Замените 'admin' на нужный логин is_admin = True if not is_admin: # Можно перенаправить на логин или показать ошибку доступа logging.warning(f"Попытка несанкционированного доступа в /admin пользователем: {session.get('user', 'Аноним')}") return redirect(url_for('login')) # Или return "Доступ запрещен", 403 # Загрузка данных для админки data = load_data() products = data.get('products', []) categories = data.get('categories', []) # Пользователи уже загружены выше для проверки админа # Обработка POST запросов из админки if request.method == 'POST': action = request.form.get('action') logging.info(f"Admin action received: {action}") try: # Обернем обработку действий в try-except для отладки 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) categories.sort() # Сортируем категории save_data(data) logging.info(f"Категория '{category_name}' добавлена.") elif not category_name: logging.warning("Попытка добавить пустую категорию.") else: logging.warning(f"Попытка добавить существующую категорию: {category_name}") return redirect(url_for('admin')) 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) # Обновляем товары, убирая удаленную категорию for product in products: if product.get('category') == category_to_delete: product['category'] = 'Без категории' # Или можно удалять ключ 'category' save_data(data) logging.info(f"Категория '{category_to_delete}' удалена.") else: logging.warning(f"Попытка удалить несуществующую или не указанную категорию: {category_to_delete}") return redirect(url_for('admin')) elif action == 'add_product': name = request.form.get('name', '').strip() price_str = request.form.get('price', '0').replace(',', '.') description = request.form.get('description', '').strip() category = request.form.get('category', 'Без категории') photos_files = request.files.getlist('photos') # Получаем цвета, удаляем пустые строки и дубликаты colors = list(dict.fromkeys(filter(None, [c.strip() for c in request.form.getlist('colors')]))) if not name or not description: # Цена может быть 0 logging.warning("Попытка добавить товар без имени или описания.") # Нужно вернуть сообщение об ошибке пользователю return redirect(url_for('admin')) # Пока просто редирект try: price = round(float(price_str), 2) if price < 0: price = 0 # Цена не может быть отрицательной except ValueError: logging.warning(f"Некорректное значение цены при добавлении товара: {price_str}") price = 0 # Цена по умолчанию photos_list = upload_photos(photos_files, name) # Используем вспомогательную функцию new_product = { 'name': name, 'price': price, # Цена уже в KGS 'description': description, 'category': category if category in categories else 'Без категории', 'photos': photos_list, 'colors': colors if colors else [] # Пустой список, если цвета не указаны } products.append(new_product) save_data(data) # Сохраняем и выгружаем на HF logging.info(f"Товар '{name}' успешно добавлен.") return redirect(url_for('admin')) elif action == 'edit_product': index = int(request.form.get('index', -1)) if 0 <= index < len(products): product_to_edit = products[index] name = request.form.get('name', '').strip() price_str = request.form.get('price', '0').replace(',', '.') description = request.form.get('description', '').strip() category = request.form.get('category', 'Без категории') photos_files = request.files.getlist('photos') # Получаем цвета, удаляем пустые строки и дубликаты colors = list(dict.fromkeys(filter(None, [c.strip() for c in request.form.getlist('colors')]))) if not name or not description: logging.warning(f"Попытка редактирования товара (индекс {index}) с пустым именем или описанием.") return redirect(url_for('admin')) try: price = round(float(price_str), 2) if price < 0: price = 0 except ValueError: logging.warning(f"Некорректное значение цены при редактировании товара {name}: {price_str}") price = product_to_edit.get('price', 0) # Оставляем старую цену # Обработка фото: если загружены новые, заменяем старые if photos_files and any(f.filename for f in photos_files): # Удаляем старые фото с HF (опционально, может быть сложно и рискованно) # delete_photos_from_hf(product_to_edit.get('photos', [])) new_photos_list = upload_photos(photos_files, name) product_to_edit['photos'] = new_photos_list logging.info(f"Фотографии для товара '{name}' обновлены.") # Если новые фото не загружены, старые остаются # Обновляем данные товара product_to_edit['name'] = name product_to_edit['price'] = price product_to_edit['description'] = description product_to_edit['category'] = category if category in categories else 'Без категории' product_to_edit['colors'] = colors if colors else [] save_data(data) logging.info(f"Товар '{name}' (индекс {index}) успешно отредактирован.") else: logging.error(f"Попытка редактирования несуществующего товара с индексом {index}.") return redirect(url_for('admin')) elif action == 'delete_product': index = int(request.form.get('index', -1)) if 0 <= index < len(products): deleted_product = products.pop(index) # Удаляем из списка и получаем удаленный элемент # Опционально: Удалить фото с HF # delete_photos_from_hf(deleted_product.get('photos', [])) save_data(data) logging.info(f"Товар '{deleted_product.get('name', 'Без имени')}' (индекс {index}) удален.") else: logging.error(f"Попытка удаления несуществующего товара с индексом {index}.") return redirect(url_for('admin')) # --- Управление пользователями --- elif action == 'add_user': login = request.form.get('login', '').strip() password = request.form.get('password', '').strip() first_name = request.form.get('first_name', '').strip() last_name = request.form.get('last_name', '').strip() country = request.form.get('country', '').strip() city = request.form.get('city', '').strip() if not login or not password or not first_name: logging.warning("Попытка добавить пользователя с неполными данными (логин, пароль, имя обязательны).") # Нужно вернуть сообщение об ошибке return redirect(url_for('admin')) if login in users: logging.warning(f"Попытка добавить пользователя с существующим логином: {login}") # Нужно вернуть сообщение об ошибке return redirect(url_for('admin')) users[login] = { 'password': password, # В реальном приложении пароль нужно хешировать! 'first_name': first_name, 'last_name': last_name, 'country': country, 'city': city, 'status': 'active' # Статус по умолчанию # Тип покупки убран } save_users(users) # Сохраняем и выгружаем на HF logging.info(f"Пользователь '{login}' успешно добавлен.") return redirect(url_for('admin')) elif action == 'delete_user': login_to_delete = request.form.get('login') if login_to_delete == 'admin': # Защита от удаления основного админа logging.warning("Попытка удаления основного администратора.") return redirect(url_for('admin')) if login_to_delete and login_to_delete in users: del users[login_to_delete] save_users(users) logging.info(f"Пользователь '{login_to_delete}' удален.") else: logging.warning(f"Попытка удалить несуществующего пользователя: {login_to_delete}") return redirect(url_for('admin')) # Курс валют убран # elif action == 'set_exchange_rate': ... except Exception as e: logging.error(f"Ошибка при выполнении действия '{action}' в админ-панели: {e}", exc_info=True) # Можно добавить сообщение об ошибке для пользователя return redirect(url_for('admin')) # Отображение админ-панели (GET запрос или после POST редиректа) admin_html = ''' Админ-панель - Soola Cosmetics

Админ-панель Soola Cosmetics

Перейти в каталог

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

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

Список существующих товаров ({{ products|length }})

{% for product in products %}

{{ product.get('name', 'Без названия') }}

ID (индекс): {{ loop.index0 }}

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

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

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

Варианты: {{ (product.get('colors')|join(', ')) if product.get('colors') else 'Нет вариантов' }}

{% if product.get('photos') %}
{% for photo in product['photos'] %} Фото {{ product.get('name', '') }} {% endfor %}
{% endif %}
Редактировать
{% for color in product.get('colors', []) %}
{% endfor %} {% if not product.get('colors') %} {# Добавляем пустое поле, если цветов нет #}
{% endif %}
{% else %}

Нет товаров для отображения.

{% endfor %}

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

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

Список существующих категорий ({{ categories|length }})

{% if categories %}
{% for category in categories %}

{{ category }}

{# Передаем имя для удаления #}
{% endfor %}
{% else %}

Нет созданных категорий.

{% endif %}

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

Добавить нового пользователя

Список зарегистрированных пользователей ({{ users|length }})

{% for login, user_info in users.items() %}

Логин: {{ login }}

Имя: {{ user_info.get('first_name', 'N/A') }} {{ user_info.get('last_name', '') }}

Страна: {{ user_info.get('country', 'Не указана') }}

Город: {{ user_info.get('city', 'Не указан') }}

Статус: {{ user_info.get('status', 'active') }}

{# Добавить хеширование пароля перед отображением или не отображать его вовсе #} {#

Пароль: {{ user_info.get('password', 'N/A') }}

#}
{% if login != 'admin' %} {# Защита от удаления основного админа #}
{# Можно добавить кнопку для блокировки/разблокировки пользователя #} {#
#} {% else %}

(Основной администратор)

{% endif %}
{% else %}

Нет зарегистрированных пользователей (кроме администратора).

{% endfor %}

Управление базой данных (Hugging Face)

Примечание: Скачивание перезапишет ваши локальные файлы {{ sync_files_list|join(', ') }} версиями с Hugging Face. Загрузка отправит ваши текущие локальные версии на Hugging Face. Автоматическое резервное копирование происходит периодически.

''' # Передаем данные в шаблон админки return render_template_string( admin_html, products=products, categories=categories, users=users, repo_id=REPO_ID, currency_code=CURRENCY_CODE, sync_files_list=SYNC_FILES # Передаем список синхронизируемых файлов # kgs_to_usd и convert_price убраны ) def upload_photos(files, product_name): """Загружает фото на HF и возвращает список имен файлов.""" photos_list = [] if not HF_TOKEN_WRITE: logging.warning("HF_TOKEN (токен для записи) не установлен. Загрузка фото отключена.") return photos_list uploads_dir = 'uploads_temp' # Временная папка для загрузки os.makedirs(uploads_dir, exist_ok=True) api = HfApi() for photo in files: if photo and photo.filename: try: # Создаем безопасное имя файла original_filename = secure_filename(photo.filename) # Добавляем временную метку для уникальности на всякий случай timestamp = datetime.now().strftime("%Y%m%d%H%M%S%f") unique_filename = f"{timestamp}_{original_filename}" temp_path = os.path.join(uploads_dir, unique_filename) photo.save(temp_path) logging.info(f"Фото временно сохранено: {temp_path}") # Загружаем на Hugging Face path_in_repo = f"photos/{unique_filename}" # Путь в репозитории HF api.upload_file( path_or_fileobj=temp_path, path_in_repo=path_in_repo, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Добавлено/обновлено фото для товара '{product_name}': {unique_filename}" ) photos_list.append(unique_filename) # Сохраняем только имя файла logging.info(f"Фото {unique_filename} успешно загружено в {REPO_ID} как {path_in_repo}") # Удаляем временный файл после успешной загрузки if os.path.exists(temp_path): os.remove(temp_path) logging.info(f"Временный файл удален: {temp_path}") except Exception as e: logging.error(f"Ошибка при обработке или загрузке фото {photo.filename}: {e}") # Пытаемся удалить временный файл, если он создался if 'temp_path' in locals() and os.path.exists(temp_path): try: os.remove(temp_path) except Exception as remove_e: logging.error(f"Не удалось удалить временный файл {temp_path} после ошибки: {remove_e}") # Очистить временную папку (если она пуста) try: if not os.listdir(uploads_dir): os.rmdir(uploads_dir) except OSError as e: logging.warning(f"Не удалось удалить временную папку {uploads_dir}: {e}") return photos_list # --- Функции Backup/Download --- @app.route('/backup', methods=['POST']) def backup(): """Запускает принудительную загрузку данных на HF.""" # Добавить проверку прав администратора? if 'user' not in session or session['user'] != 'admin': return "Доступ запрещен", 403 try: logging.info("Запущена принудительная загрузка на Hugging Face...") upload_db_to_hf() logging.info("Принудительная загрузка на Hugging Face завершена.") # Можно вернуть JSON ответ для JS return "Резервная копия успешно инициирована.", 200 except Exception as e: logging.error(f"Ошибка во время ручного бэкапа: {e}") return f"Ошибка во время создания резервной копии: {e}", 500 @app.route('/download', methods=['GET']) def download(): """Запускает принудительное скачивание данных с HF.""" # Добавить проверку прав администратора? if 'user' not in session or session['user'] != 'admin': return "Доступ запрещен", 403 try: logging.info("Запущено принудительное скачивание с Hugging Face...") download_db_from_hf() logging.info("Принудительное скачивание с Hugging Face завершено.") # Важно: После скачивания приложение может потребовать перезапуска, # чтобы подхватить новые данные, если они загружаются в память при старте. # Flask в режиме debug обычно перезапускается сам. # В продакшене может потребоваться ручной перезапуск или сигнал серверу. return "База данных успешно скачана с Hugging Face. Может потребоваться перезапуск для применения.", 200 except RepositoryNotFoundError: return f"Ошибка: Репозиторий {REPO_ID} не найден на Hugging Face.", 404 except Exception as e: logging.error(f"Ошибка во время ручного скачивания: {e}") return f"Ошибка во время скачивания базы данных: {e}", 500 # --- Запуск приложения --- if __name__ == '__main__': # Проверяем наличие необходимых токенов if not HF_TOKEN_WRITE: logging.warning("Переменная окружения HF_TOKEN (для записи) не установлена!") if not HF_TOKEN_READ: logging.warning("Переменная окружения HF_TOKEN_READ (для чтения) не установлена! Будет использоваться HF_TOKEN.") # Если токен для чтения не задан, пытаемся использовать токен для записи (часто он имеет права и на чтение) HF_TOKEN_READ = HF_TOKEN_WRITE if not HF_TOKEN_READ: logging.error("Токены для чтения и записи на Hugging Face не установлены! Синхронизация не будет работать.") # Попытка первоначальной загрузки данных при старте logging.info("Первоначальная загрузка данных при старте приложения...") try: # Сначала пытаемся скачать последнюю версию с HF download_db_from_hf() except RepositoryNotFoundError: logging.error(f"*** Репозиторий {REPO_ID} не найден на Hugging Face! Приложение будет использовать локальные файлы (если есть) или создаст новые. ***") except Exception as e: logging.error(f"*** Ошибка при первоначальном скачивании данных с Hugging Face: {e}. Будут использованы локальные файлы. ***") # Загружаем данные в память (или создаем пустые структуры, если файлов нет) load_data() load_users() # Запускаем поток для периодического резервного копирования, только если есть токен для записи if HF_TOKEN_WRITE: backup_thread = threading.Thread(target=periodic_backup, daemon=True) backup_thread.start() logging.info("Поток периодического резервного копирования запущен.") else: logging.warning("Поток периодического резервного копирования НЕ запущен из-за отсутствия HF_TOKEN.") # Запуск Flask приложения # host='0.0.0.0' делает приложение доступным извне Docker контейнера или в локальной сети # debug=True удобно для разработки, но НЕ ИСПОЛЬЗУЙТЕ В ПРОДАКШЕНЕ! # port=7860 - стандартный порт для Gradio/HF Spaces, можно изменить logging.info("Запуск Flask приложения...") app.run(debug=True, host='0.0.0.0', port=7860)