from flask import Flask, render_template_string, request, redirect, url_for, send_file, flash, jsonify 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, HfHubHTTPError from werkzeug.utils import secure_filename from dotenv import load_dotenv import requests import uuid load_dotenv() app = Flask(__name__) app.secret_key = 'your_unique_secret_key_soola_cosmetics_67890_no_login' DATA_FILE = 'data.json' SYNC_FILES = [DATA_FILE] REPO_ID = "Kgshop/altynroller" HF_TOKEN_WRITE = os.getenv("HF_TOKEN") HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") STORE_ADDRESSES = [ "ТЦ Новый Ялян: 2 этаж, сектор G, бутик 50-52", ] CURRENCY_CODE = 'KZT' CURRENCY_NAME = 'Казахстанский тенге' DOWNLOAD_RETRIES = 3 DOWNLOAD_DELAY = 5 STATUS_MAP_RU = { "new": "Новый", "accepted": "Принят", "prepared": "Собран", "shipped": "Отправлен" } 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 logging.info(f"Attempting download for {files_to_download} from {REPO_ID}...") all_successful = True for file_name in files_to_download: success = False for attempt in range(retries + 1): try: logging.info(f"Downloading {file_name} (Attempt {attempt + 1}/{retries + 1})...") 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 ) logging.info(f"Successfully downloaded {file_name} to {local_path}.") success = True break except RepositoryNotFoundError: logging.error(f"Repository {REPO_ID} not found. Download cancelled for all files.") return False except HfHubHTTPError as e: if e.response.status_code == 404: logging.warning(f"File {file_name} not found in repo {REPO_ID} (404). Skipping this file.") 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': {}}, f) logging.info(f"Created empty local file {file_name} because it was not found on HF.") except Exception as create_e: logging.error(f"Failed to create empty local file {file_name}: {create_e}") success = False break else: logging.error(f"HTTP error downloading {file_name} (Attempt {attempt + 1}): {e}. Retrying in {delay}s...") except requests.exceptions.RequestException as e: logging.error(f"Network error downloading {file_name} (Attempt {attempt + 1}): {e}. Retrying in {delay}s...") except Exception as e: logging.error(f"Unexpected error downloading {file_name} (Attempt {attempt + 1}): {e}. Retrying in {delay}s...", exc_info=True) if attempt < retries: time.sleep(delay) if not success: logging.error(f"Failed to download {file_name} after {retries + 1} attempts.") all_successful = False logging.info(f"Download process finished. Overall success: {all_successful}") return all_successful def upload_db_to_hf(specific_file=None): if not HF_TOKEN_WRITE: logging.warning("HF_TOKEN (for writing) not set. Skipping upload to Hugging Face.") return try: api = HfApi() files_to_upload = [specific_file] if specific_file else SYNC_FILES logging.info(f"Starting upload of {files_to_upload} to HF repo {REPO_ID}...") 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')}" ) logging.info(f"File {file_name} successfully uploaded to Hugging Face.") except Exception as e: logging.error(f"Error uploading file {file_name} to Hugging Face: {e}") else: logging.warning(f"File {file_name} not found locally, skipping upload.") logging.info("Finished uploading files to HF.") except Exception as e: logging.error(f"General error during Hugging Face upload initialization or process: {e}", exc_info=True) def periodic_backup(): backup_interval = 1800 logging.info(f"Setting up periodic backup every {backup_interval} seconds.") while True: time.sleep(backup_interval) logging.info("Starting periodic backup...") upload_db_to_hf() logging.info("Periodic backup finished.") def load_data(): default_data = {'products': [], 'categories': [], 'orders': {}} try: with open(DATA_FILE, 'r', encoding='utf-8') as file: data = json.load(file) logging.info(f"Local data loaded successfully from {DATA_FILE}") if not isinstance(data, dict): logging.warning(f"Local {DATA_FILE} is not a dictionary. Attempting download.") raise FileNotFoundError if 'products' not in data: data['products'] = [] if 'categories' not in data: data['categories'] = [] if 'orders' not in data: data['orders'] = {} return data except FileNotFoundError: logging.warning(f"Local file {DATA_FILE} not found. Attempting download from HF.") except json.JSONDecodeError: logging.error(f"Error decoding JSON in local {DATA_FILE}. File might be corrupt. Attempting download.") if download_db_from_hf(specific_file=DATA_FILE): try: with open(DATA_FILE, 'r', encoding='utf-8') as file: data = json.load(file) logging.info(f"Data loaded successfully from {DATA_FILE} after download.") if not isinstance(data, dict): logging.error(f"Downloaded {DATA_FILE} is not a dictionary. Using default.") return default_data if 'products' not in data: data['products'] = [] if 'categories' not in data: data['categories'] = [] if 'orders' not in data: data['orders'] = {} return data except FileNotFoundError: logging.error(f"File {DATA_FILE} still not found even after download reported success. Using default.") return default_data except json.JSONDecodeError: logging.error(f"Error decoding JSON in downloaded {DATA_FILE}. Using default.") return default_data except Exception as e: logging.error(f"Unknown error loading downloaded {DATA_FILE}: {e}. Using default.", exc_info=True) return default_data else: logging.error(f"Failed to download {DATA_FILE} from HF after retries. Using empty default data structure.") if not os.path.exists(DATA_FILE): try: with open(DATA_FILE, 'w', encoding='utf-8') as f: json.dump(default_data, f) logging.info(f"Created empty local file {DATA_FILE} after failed download.") except Exception as create_e: logging.error(f"Failed to create empty local file {DATA_FILE}: {create_e}") return default_data def save_data(data): try: if not isinstance(data, dict): logging.error("Attempted to save invalid data structure (not a dict). Aborting save.") return if 'products' not in data: data['products'] = [] if 'categories' not in data: data['categories'] = [] if 'orders' not in data: data['orders'] = {} with open(DATA_FILE, 'w', encoding='utf-8') as file: json.dump(data, file, ensure_ascii=False, indent=4) logging.info(f"Data successfully saved to {DATA_FILE}") upload_db_to_hf(specific_file=DATA_FILE) except Exception as e: logging.error(f"Error saving data to {DATA_FILE}: {e}", exc_info=True) CATALOG_TEMPLATE = ''' ROLLER ALTYN - Каталог
ROLLER ALTYN Logo

ROLLER ALTYN

Наши адреса в г. Алматы

{% for address in store_addresses %}

{{ address }}

{% endfor %}
{% for category in categories %} {% endfor %}
{% for product in products %}
{% if product.get('is_top', False) %} Топ {% endif %}
{% if product.get('photos') and product['photos']|length > 0 %} {{ product['name'] }} {% else %} No Image {% endif %}

{{ product['name'] }}

{% if product.get('variants') and product.variants|length > 0 %} от {{ "%.2f"|format(product.variants|map(attribute='price')|min) }} {{ currency_code }} {% else %} - {% endif %}
{% endfor %} {% if not products %}

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

{% endif %}
Связаться по WhatsApp
''' 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 %}

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

{% if product.get('variants') and product.variants|length > 0 %}

Цена: от {{ "%.2f"|format(product.variants|map(attribute='price')|min) }} {{ currency_code }}

Доступные варианты:

{% endif %}

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

''' ORDER_TEMPLATE = ''' Заказ №{{ order.id }} - ROLLER ALTYN
{% if order %}

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

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

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

{% for item in order.cart %}
{{ item.name }}
{{ item.name }} Вариант: {{ item.variantName }} {{ "%.2f"|format(item.price) }} {{ currency_code }} × {{ item.quantity }}
{{ "%.2f"|format(item.price * item.quantity) }} {{ currency_code }}
{% endfor %}

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

Статус и подтверждение

Текущий статус: {{ status_map_ru.get(order.status, order.status) }}

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

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

Ошибка

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

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

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

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

История заказов

{% if orders %} {% for order in orders | sort(attribute='created_at', reverse=true) %} {% set current_status = order.get('status', 'new') %}
Заказ №{{ order.id }} {{ status_map_ru.get(current_status, current_status) }} Дата: {{ order.created_at }} | Итого: {{ "%.2f"|format(order.total_price) }} {{ currency_code }}

Состав заказа:

Просмотр
{% endfor %} {% else %}

Активных заказов нет.

{% endif %}

Категории

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

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

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

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

{% endif %}

Инфо о магазине

Адреса в г. Алматы:

    {% for address in store_addresses %}
  • {{ address }}
    • {% endfor %}

Валюта: {{ currency_name }} ({{ currency_code }})

Товары

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

Варианты и цены *:


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

{% if products %}
{% for product in products %}
{% if product.get('photos') and product.photos|length > 0 %} Фото {% else %} Нет фото {% endif %}

{{ product['name'] }} {% if product.get('is_top', False) %} Топ{% endif %}

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

Варианты: {{ product.get('variants', [])|length }} шт.

{% endfor %}
{% else %}

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

{% endif %}
''' @app.route('/') def catalog(): data = load_data() all_products = data.get('products', []) categories = sorted(data.get('categories', [])) products_in_stock = [p for p in all_products if p.get('variants')] for p in products_in_stock: p['variants'] = sorted(p.get('variants', []), key=lambda v: v.get('price', 0)) products_sorted = sorted(products_in_stock, key=lambda p: (not p.get('is_top', False), p.get('name', '').lower())) return render_template_string( CATALOG_TEMPLATE, products=products_sorted, categories=categories, repo_id=REPO_ID, store_addresses=STORE_ADDRESSES, currency_code=CURRENCY_CODE ) @app.route('/product/') def product_detail(index): data = load_data() all_products = data.get('products', []) products_in_stock = [p for p in all_products if p.get('variants')] for p in products_in_stock: p['variants'] = sorted(p.get('variants', []), key=lambda v: v.get('price', 0)) 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, products_index=index, 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 = [] data = load_data() products_map = {p['id']: p for p in data.get('products', [])} for item in cart_items: product = products_map.get(item.get('productId')) if not product: continue variant_exists = any(v['name'] == item.get('variantName') and v['price'] == item.get('price') for v in product.get('variants', [])) if not variant_exists: continue try: price = float(item['price']) quantity = int(item['quantity']) if price < 0 or quantity <= 0: raise ValueError("Invalid price/qty") total_price += price * quantity photo_name = item.get('photo') photo_url = f"https://huggingface.co/datasets/{REPO_ID}/resolve/main/photos/{photo_name}" if photo_name else "https://via.placeholder.com/60x60.png?text=N/A" processed_cart.append({ "name": item['name'], "price": price, "quantity": quantity, "variantName": item.get('variantName', 'N/A'), "photo": photo_name, "photo_url": photo_url }) except (ValueError, TypeError): return jsonify({"error": "Неверная цена или количество."}), 400 if not processed_cart: return jsonify({"error": "Не удалось обработать товары в корзине."}), 400 order_id = f"{datetime.now().strftime('%y%m%d%H%M')}-{uuid.uuid4().hex[:4]}" new_order = { "id": order_id, "created_at": datetime.now().strftime('%Y-%m-%d %H:%M:%S'), "cart": processed_cart, "total_price": round(total_price, 2), "status": "new" } try: data['orders'][order_id] = new_order save_data(data) return jsonify({"order_id": order_id}), 201 except Exception as e: logging.error(f"Failed to save order {order_id}: {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, status_map_ru=STATUS_MAP_RU, currency_code=CURRENCY_CODE) @app.route('/admin', methods=['GET', 'POST']) def admin(): data = load_data() if 'orders' not in data: data['orders'] = {} needs_save = False for p in data.get('products', []): if 'id' not in p: p['id'] = str(uuid.uuid4()) needs_save = True if needs_save: save_data(data) if request.method == 'POST': action = request.form.get('action') try: if action == 'update_order_status': order_id = request.form.get('order_id') new_status = request.form.get('new_status') if order_id in data['orders'] and new_status in STATUS_MAP_RU: data['orders'][order_id]['status'] = new_status save_data(data) flash(f"Статус заказа №{order_id} изменен.", 'success') else: flash("Ошибка обновления статуса заказа.", 'error') elif action == 'add_category': category_name = request.form.get('category_name', '').strip() if category_name and category_name not in data['categories']: data['categories'].append(category_name) save_data(data) flash(f"Категория '{category_name}' добавлена.", 'success') else: flash("Ошибка: Категория пуста или уже существует.", 'error') elif action == 'delete_category': category_to_delete = request.form.get('category_name') if category_to_delete in data['categories']: data['categories'].remove(category_to_delete) for product in data['products']: if product.get('category') == category_to_delete: product['category'] = 'Без категории' save_data(data) flash(f"Категория '{category_to_delete}' удалена.", 'success') else: flash("Категория не найдена.", 'error') elif action == 'add_product' or action == 'edit_product': name = request.form.get('name', '').strip() if not name: flash("Название товара обязательно.", 'error') return redirect(url_for('admin')) variant_names = [v.strip() for v in request.form.getlist('variant_names') if v.strip()] variant_prices_str = request.form.getlist('variant_prices') if not variant_names or not variant_prices_str or len(variant_names) != len(variant_prices_str): flash("Ошибка в вариантах. У каждого варианта должно быть название и цена.", 'error') return redirect(url_for('admin')) variants = [] try: for i, v_name in enumerate(variant_names): price = round(float(variant_prices_str[i].replace(',', '.')), 2) if price < 0: price = 0 variants.append({'name': v_name, 'price': price}) except ValueError: flash("Неверный формат цены в вариантах.", 'error') return redirect(url_for('admin')) product_data = { 'name': name, 'description': request.form.get('description', '').strip(), 'category': request.form.get('category'), 'is_top': 'is_top' in request.form, 'variants': variants } newly_uploaded_photos = [] photos_files = request.files.getlist('photos') if photos_files and any(f.filename for f in photos_files): if not HF_TOKEN_WRITE: flash("Токен HF_TOKEN не настроен, фото не загружены.", "warning") else: api = HfApi() uploads_dir = 'uploads_temp' os.makedirs(uploads_dir, exist_ok=True) for photo in photos_files[:10]: if photo and photo.filename: try: ext = os.path.splitext(photo.filename)[1].lower() if ext not in ['.jpg', '.jpeg', '.png', '.gif', '.webp']: continue safe_name = secure_filename(name.replace(' ', '_'))[:50] photo_filename = f"{safe_name}_{datetime.now().strftime('%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"Photo for {name}" ) newly_uploaded_photos.append(photo_filename) os.remove(temp_path) except Exception as e: logging.error(f"Error uploading photo {photo.filename}: {e}") flash(f"Ошибка загрузки фото {photo.filename}.", 'error') if action == 'add_product': product_data['id'] = str(uuid.uuid4()) product_data['photos'] = newly_uploaded_photos data['products'].append(product_data) flash(f"Товар '{name}' добавлен.", 'success') else: # edit_product product_id = request.form.get('product_id') product_index = next((i for i, p in enumerate(data['products']) if p.get('id') == product_id), -1) if product_index != -1: product_to_edit = data['products'][product_index] current_photos = product_to_edit.get('photos', []) photos_to_delete = request.form.getlist('delete_photos') if photos_to_delete and HF_TOKEN_WRITE: try: HfApi().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 ) flash(f"Удалено {len(photos_to_delete)} фото.", 'success') except Exception as e: logging.error(f"Error deleting photos from HF: {e}") flash("Не удалось удалить фото с сервера.", "warning") updated_photos = [p for p in current_photos if p not in photos_to_delete] product_data['photos'] = updated_photos + newly_uploaded_photos product_data['id'] = product_id data['products'][product_index] = product_data flash(f"Товар '{name}' обновлен.", 'success') else: flash("Товар для редактирования не найден.", 'error') save_data(data) elif action == 'delete_product': product_id = request.form.get('product_id') product_to_delete = next((p for p in data['products'] if p.get('id') == product_id), None) if product_to_delete: data['products'].remove(product_to_delete) photos_to_delete = product_to_delete.get('photos', []) if photos_to_delete and HF_TOKEN_WRITE: try: HfApi().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 ) except Exception as e: logging.error(f"Error deleting photos from HF: {e}") flash("Не удалось удалить фото с сервера.", "warning") save_data(data) flash(f"Товар '{product_to_delete.get('name')}' удален.", 'success') else: flash("Товар для удаления не найден.", 'error') return redirect(url_for('admin')) except Exception as e: logging.error(f"Error processing admin action '{action}': {e}", exc_info=True) flash(f"Произошла внутренняя ошибка: {e}", 'error') return redirect(url_for('admin')) current_data = load_data() return render_template_string( ADMIN_TEMPLATE, products=sorted(current_data.get('products', []), key=lambda p: p.get('name', '').lower()), categories=sorted(current_data.get('categories', [])), orders=list(current_data.get('orders', {}).values()), status_map_ru=STATUS_MAP_RU, repo_id=REPO_ID, store_addresses=STORE_ADDRESSES, currency_code=CURRENCY_CODE, currency_name=CURRENCY_NAME ) if __name__ == '__main__': logging.info("Application starting up...") download_db_from_hf() load_data() if HF_TOKEN_WRITE: threading.Thread(target=periodic_backup, daemon=True).start() port = int(os.environ.get('PORT', 7860)) app.run(debug=False, host='0.0.0.0', port=port)