from flask import Flask, render_template_string, request, redirect, url_for import json import os import threading import time import uuid from datetime import datetime import zoneinfo from huggingface_hub import HfApi, hf_hub_download app = Flask(__name__) DATA_FILE = 'data_fabrics.json' REPO_ID = "Kgshop/vahid" HF_TOKEN_WRITE = os.getenv("HF_TOKEN") HF_TOKEN_READ = os.getenv("HF_TOKEN_READ") INITIALIZED = False def get_almaty_time(): return datetime.now(zoneinfo.ZoneInfo("Asia/Almaty")).strftime('%d.%m.%Y %H:%M:%S') def load_data(): global INITIALIZED if not INITIALIZED: try: hf_hub_download( repo_id=REPO_ID, filename=DATA_FILE, repo_type="dataset", token=HF_TOKEN_READ, local_dir=".", local_dir_use_symlinks=False ) except Exception: pass INITIALIZED = True try: with open(DATA_FILE, 'r', encoding='utf-8') as file: data = json.load(file) if not isinstance(data, dict): data = {'products': [], 'history': []} if 'settings' not in data: data['settings'] = { 'warehouses': [{'id': 'w_main', 'name': 'Основной склад'}], 'shops': [{'id': 's_main', 'name': 'Основной магазин'}], 'employees': [] } if 'products' not in data: data['products'] = [] if 'history' not in data: data['history'] = [] for p in data['products']: if 'price_retail' not in p: p['price_retail'] = p.get('price', 0) if 'price_wholesale' not in p: p['price_wholesale'] = p.get('price', 0) if 'variants' not in p: p['variants'] = [] if 'photo' not in p: p['photo'] = '' if 'box_qty' not in p: p['box_qty'] = 1 if 'stocks' not in p: p['stocks'] = {} if 'colors' in p: for c in p['colors']: for loc, amt in c.get('stocks', {}).items(): p['stocks'][loc] = p['stocks'].get(loc, 0) + amt del p['colors'] return data except Exception: return { 'settings': { 'warehouses': [{'id': 'w_main', 'name': 'Основной склад'}], 'shops': [{'id': 's_main', 'name': 'Основной магазин'}], 'employees': [] }, 'products': [], 'history': [] } def save_data(data): try: with open(DATA_FILE, 'w', encoding='utf-8') as file: json.dump(data, file, ensure_ascii=False, indent=4) threading.Thread(target=upload_db_to_hf, daemon=True).start() except Exception: pass def upload_db_to_hf(): try: api = HfApi() api.upload_file( path_or_fileobj=DATA_FILE, path_in_repo=DATA_FILE, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN_WRITE, commit_message=f"Backup {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ) except Exception: pass def periodic_backup(): while True: upload_db_to_hf() time.sleep(800) def add_history(data, action, product, amount, price, from_loc, to_loc, emp_name): data['history'].insert(0, { 'time': get_almaty_time(), 'action': action, 'product': product, 'amount': amount, 'price': price, 'from_loc_name': from_loc, 'to_loc_name': to_loc, 'emp_name': emp_name }) data['history'] = data['history'][:1000] def get_loc_name(data, loc_id): for w in data['settings']['warehouses']: if w['id'] == loc_id: return w['name'] for s in data['settings']['shops']: if s['id'] == loc_id: return s['name'] return loc_id def handle_post(req, data, role, emp): action = req.form.get('action') emp_name = emp['name'] if emp else 'Админ' if action == 'add' and role == 'admin': loc_id = req.form.get('loc_id') try: price_retail = int(req.form.get('price_retail', 0)) except ValueError: price_retail = 0 try: price_wholesale = int(req.form.get('price_wholesale', 0)) except ValueError: price_wholesale = 0 photo = req.form.get('photo', '') try: rolls = int(req.form.get('rolls', 0)) except ValueError: rolls = 0 try: box_qty = int(req.form.get('box_qty', 1)) except ValueError: box_qty = 1 v_names = req.form.getlist('v_name[]') v_prices = req.form.getlist('v_price[]') variants = [] for n, pr in zip(v_names, v_prices): if n.strip(): try: variants.append({'name': n.strip(), 'price': int(pr)}) except ValueError: pass data['products'].insert(0, { 'id': str(time.time()), 'name': req.form['name'], 'description': req.form['description'], 'price_retail': price_retail, 'price_wholesale': price_wholesale, 'variants': variants, 'photo': photo, 'box_qty': box_qty, 'stocks': {loc_id: rolls} }) elif action == 'edit' and role == 'admin': idx = int(req.form['index']) if 0 <= idx < len(data['products']): p = data['products'][idx] p['name'] = req.form.get('name', p['name']) p['description'] = req.form.get('description', p.get('description', '')) try: p['price_retail'] = int(req.form.get('price_retail', p.get('price_retail', 0))) except ValueError: pass try: p['price_wholesale'] = int(req.form.get('price_wholesale', p.get('price_wholesale', 0))) except ValueError: pass try: p['box_qty'] = int(req.form.get('box_qty', p.get('box_qty', 1))) except ValueError: pass v_names = req.form.getlist('v_name[]') v_prices = req.form.getlist('v_price[]') variants = [] for n, pr in zip(v_names, v_prices): if n.strip(): try: variants.append({'name': n.strip(), 'price': int(pr)}) except ValueError: pass p['variants'] = variants photo = req.form.get('photo', '') if photo: p['photo'] = photo elif action == 'delete' and role == 'admin': idx = int(req.form['index']) if 0 <= idx < len(data['products']): del_name = data['products'][idx]['name'] data['products'].pop(idx) add_history(data, 'Удаление', del_name, 0, 0, '', '', emp_name) elif action == 'income' and role == 'admin': idx = int(req.form['product_index']) amt = int(req.form['rolls']) loc_id = req.form['loc_id'] p = data['products'][idx] p['stocks'][loc_id] = p.get('stocks', {}).get(loc_id, 0) + amt add_history(data, 'Приход', p['name'], amt, p.get('price_wholesale', 0), '', get_loc_name(data, loc_id), emp_name) elif action == 'transfer' and role == 'admin': idx = int(req.form['pIndex']) amt = int(req.form['amt']) from_loc = req.form['from_loc'] to_loc = req.form['to_loc'] p = data['products'][idx] av = p.get('stocks', {}).get(from_loc, 0) if amt > av: amt = av if amt > 0: p['stocks'][from_loc] -= amt p['stocks'][to_loc] = p['stocks'].get(to_loc, 0) + amt from_name = get_loc_name(data, from_loc) to_name = get_loc_name(data, to_loc) add_history(data, 'Перемещение', p['name'], amt, 0, from_name, to_name, emp_name) elif action == 'checkout': cart_data = req.form.get('cart', '[]') cart = json.loads(cart_data) for item in cart: idx = item.get('pIndex') amt = item.get('amt') loc = item.get('loc') pName = item.get('pName') itemPrice = item.get('price', 0) if 0 <= idx < len(data['products']): p = data['products'][idx] av = p.get('stocks', {}).get(loc, 0) if amt > av: amt = av if amt > 0: p['stocks'][loc] -= amt loc_name = get_loc_name(data, loc) add_history(data, 'Продажа', pName, amt, itemPrice, loc_name, '', emp_name) elif action == 'add_loc' and role == 'admin': l_type = req.form['l_type'] l_name = req.form['l_name'] l_id = f"loc_{int(time.time()*1000)}" data['settings'][l_type].append({'id': l_id, 'name': l_name}) elif action == 'del_loc' and role == 'admin': l_type = req.form['l_type'] l_id = req.form['l_id'] data['settings'][l_type] = [x for x in data['settings'][l_type] if x['id'] != l_id] elif action == 'add_emp' and role == 'admin': e_name = req.form['e_name'] e_shop = req.form['e_shop'] e_id = str(uuid.uuid4())[:8] data['settings']['employees'].append({'id': e_id, 'name': e_name, 'token': e_id, 'shop_id': e_shop}) elif action == 'del_emp' and role == 'admin': e_id = req.form['e_id'] data['settings']['employees'] = [x for x in data['settings']['employees'] if x['id'] != e_id] MAIN_HTML = ''' Учет товаров

Учет товаров

{% if role == 'admin' %}
{% elif role == 'employee' %}
{% endif %}
Корзина (0)
''' @app.route('/') def public_index(): return redirect(url_for('public_retail')) @app.route('/catalog/retail') def public_retail(): data = load_data() return render_template_string(MAIN_HTML, data=data, role='public', emp=None, price_type='retail') @app.route('/catalog/wholesale') def public_wholesale(): data = load_data() return render_template_string(MAIN_HTML, data=data, role='public', emp=None, price_type='wholesale') @app.route('/admin', methods=['GET', 'POST']) def admin_panel(): data = load_data() if request.method == 'POST': handle_post(request, data, 'admin', None) save_data(data) return redirect(url_for('admin_panel')) return render_template_string(MAIN_HTML, data=data, role='admin', emp=None, price_type='both') @app.route('/emp/', methods=['GET', 'POST']) def employee_panel(token): data = load_data() emp = next((e for e in data['settings']['employees'] if e.get('token') == token), None) if not emp: return "Неверная ссылка", 403 if request.method == 'POST': handle_post(request, data, 'employee', emp) save_data(data) return redirect(f'/emp/{token}') return render_template_string(MAIN_HTML, data=data, role='employee', emp=emp, price_type='both') if __name__ == '__main__': load_data() backup_thread = threading.Thread(target=periodic_backup, daemon=True) backup_thread.start() app.run(debug=True, host='0.0.0.0', port=7860)