#!/usr/bin/env python3
import os
from flask import Flask, request, Response, render_template_string, jsonify, redirect, url_for
import hmac
import hashlib
import json
from urllib.parse import unquote, parse_qs, quote
import time
from datetime import datetime
import threading
import random
import re
import pytz
from huggingface_hub import HfApi, hf_hub_download
from huggingface_hub.utils import RepositoryNotFoundError
BOT_TOKEN = os.getenv("BOT_TOKEN", "7835463659:AAGNePbelZIAOeaglyQi1qulOqnjs4BGQn4")
HOST = '0.0.0.0'
PORT = 7860
DATA_FILE = 'data.json'
ORG_INFO_FILE = 'organization_info.json'
REPO_ID = "flpolprojects/examplebonus"
HF_DATA_FILE_PATH = "data.json"
HF_ORG_INFO_FILE_PATH = "organization_info.json"
HF_TOKEN_WRITE = os.getenv("HF_TOKEN_WRITE")
HF_TOKEN_READ = os.getenv("HF_TOKEN_READ")
BISHKEK_TZ = pytz.timezone('Asia/Bishkek')
app = Flask(__name__)
app.secret_key = os.urandom(24)
_data_lock = threading.Lock()
_org_info_lock = threading.Lock()
visitor_data_cache = {}
org_info_cache = {}
def generate_unique_id(all_data):
while True:
new_id = str(random.randint(10000, 99999))
if new_id not in all_data:
return new_id
def download_files_from_hf():
global visitor_data_cache, org_info_cache
if not HF_TOKEN_READ:
return
try:
hf_hub_download(repo_id=REPO_ID, filename=HF_DATA_FILE_PATH, repo_type="dataset", token=HF_TOKEN_READ, local_dir=".", local_dir_use_symlinks=False, force_download=True, etag_timeout=10)
with _data_lock:
try:
with open(DATA_FILE, 'r', encoding='utf-8') as f:
visitor_data_cache = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
visitor_data_cache = {}
except Exception:
pass
try:
hf_hub_download(repo_id=REPO_ID, filename=HF_ORG_INFO_FILE_PATH, repo_type="dataset", token=HF_TOKEN_READ, local_dir=".", local_dir_use_symlinks=False, force_download=True, etag_timeout=10)
with _org_info_lock:
try:
with open(ORG_INFO_FILE, 'r', encoding='utf-8') as f:
org_info_cache = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
org_info_cache = {}
except Exception:
pass
def load_visitor_data():
global visitor_data_cache
with _data_lock:
if not visitor_data_cache:
try:
with open(DATA_FILE, 'r', encoding='utf-8') as f:
visitor_data_cache = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
visitor_data_cache = {}
return visitor_data_cache
def save_visitor_data(data):
with _data_lock:
visitor_data_cache.update(data)
with open(DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(visitor_data_cache, f, ensure_ascii=False, indent=4)
upload_data_to_hf_async()
def load_org_info():
global org_info_cache
with _org_info_lock:
if not org_info_cache:
try:
with open(ORG_INFO_FILE, 'r', encoding='utf-8') as f:
org_info_cache = json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
org_info_cache = {
"name": "Название вашей организации",
"phones": ["+996 (555) 123-456"],
"address": "г. Бишкек, ул. Примерная, 123",
"links": [{"label": "Наш сайт", "url": "https://example.com"}]
}
return org_info_cache
def save_org_info(data):
global org_info_cache
with _org_info_lock:
org_info_cache = data
with open(ORG_INFO_FILE, 'w', encoding='utf-8') as f:
json.dump(org_info_cache, f, ensure_ascii=False, indent=4)
upload_data_to_hf_async(is_org_info=True)
def upload_data_to_hf(is_org_info=False):
if not HF_TOKEN_WRITE:
return
file_to_upload = ORG_INFO_FILE if is_org_info else DATA_FILE
path_in_repo = HF_ORG_INFO_FILE_PATH if is_org_info else HF_DATA_FILE_PATH
commit_msg = f"Update org info {datetime.now(BISHKEK_TZ).strftime('%Y-%m-%d %H:%M:%S')}" if is_org_info else f"Update bonus data {datetime.now(BISHKEK_TZ).strftime('%Y-%m-%d %H:%M:%S')}"
if not os.path.exists(file_to_upload) or os.path.getsize(file_to_upload) == 0:
return
try:
api = HfApi()
api.upload_file(
path_or_fileobj=file_to_upload,
path_in_repo=path_in_repo,
repo_id=REPO_ID,
repo_type="dataset",
token=HF_TOKEN_WRITE,
commit_message=commit_msg
)
except Exception as e:
# Silently fail for now
pass
def upload_data_to_hf_async(is_org_info=False):
upload_thread = threading.Thread(target=upload_data_to_hf, args=(is_org_info,), daemon=True)
upload_thread.start()
def periodic_backup():
if not HF_TOKEN_WRITE:
return
while True:
time.sleep(3600)
upload_data_to_hf(is_org_info=False)
time.sleep(5)
upload_data_to_hf(is_org_info=True)
def verify_telegram_data(init_data_str):
try:
parsed_data = parse_qs(init_data_str)
received_hash = parsed_data.pop('hash', [None])[0]
if not received_hash:
return None, False
data_check_list = sorted([(k, v[0]) for k, v in parsed_data.items()])
data_check_string = "\n".join([f"{k}={v}" for k, v in data_check_list])
secret_key = hmac.new("WebAppData".encode(), BOT_TOKEN.encode(), hashlib.sha256).digest()
calculated_hash = hmac.new(secret_key, data_check_string.encode(), hashlib.sha256).hexdigest()
if calculated_hash == received_hash:
return parsed_data, True
return parsed_data, False
except Exception:
return None, False
def clean_phone_number(phone_str):
return re.sub(r'\D', '', phone_str)
# =========== CLIENT-SIDE TEMPLATES ===========
MAIN_TEMPLATE = """
Bonus
Ваши бонусы
{{ "%.2f"|format(user.bonuses|float) }}
Ваш долг
{{ "%.2f"|format(user.debts|float) }}
Ваш ID клиента
{{ user.id }}
"""
HISTORY_TEMPLATE = """
История операций
‹ Назад
История операций
{% if user.combined_history %}
{% for item in user.combined_history %}
-
{{ item.description }}
{{ item.date_str }}
{% if item.transaction_type == 'bonus' %}
{{ '+' if item.type == 'accrual' else '-' }}{{ "%.2f"|format(item.amount|float) }}
{% elif item.transaction_type == 'debt' %}
{{ '+' if item.type == 'accrual' else '-' }}{{ "%.2f"|format(item.amount|float) }}
{% endif %}
{% endfor %}
{% else %}
Операций пока не было.
{% endif %}
"""
INVOICES_TEMPLATE = """
Мои накладные
‹ Назад
Мои накладные
{% if user.invoices %}
{% for invoice in user.invoices %}
{{ invoice.date_str }}
Накладная #{{ invoice.id }}
{{ "%.2f"|format(invoice.total_amount|float) }}
| Товар | Кол-во | Цена | Сумма |
{% for item in invoice.items %}
| {{ item.name }} |
{{ item.quantity }} |
{{ "%.2f"|format(item.price_per_unit|float) }} |
{{ "%.2f"|format(item.total_price|float) }} |
{% endfor %}
{% endfor %}
{% else %}
У вас пока нет накладных.
{% endif %}
"""
CARD_TEMPLATE = """
Визитка
‹ Назад
{{ org_info.name }}
Номера телефонов
{% for phone in org_info.phones %}
{% endfor %}
Адрес
{{ org_info.address }}
{% if org_info.links %}
{% endif %}
"""
# =========== ADMIN-SIDE TEMPLATE ===========
ADMIN_TEMPLATE = """
Bonus Admin
Панель администратора Bonus
{{ summary.total_users }}
Всего клиентов
{{ "%.2f"|format(summary.total_bonuses|float) }}
Всего бонусов
{{ "%.2f"|format(summary.total_debts|float) }}
Всего долгов
{{ summary.users_with_debt }}
Клиенты с долгом
{% if users %}
{% for user in users|sort(attribute='visited_at', reverse=true) %}
{{ user.first_name or '' }} {{ user.last_name or '' }}
@{{ user.username if user.username else user.phone_number }} | ID: {{ user.id }}
Бонусы
{{ "%.2f"|format(user.bonuses|float) }}
Долг
{{ "%.2f"|format(user.debts|float if user.debts else 0) }}
{% if user.telegram_id == None %}{% endif %}
{% endfor %}
{% else %}
Пользователей пока нет.
{% endif %}
"""
@app.route('/')
def index():
page = request.args.get('page', 'home')
user_id_str = request.args.get('user_id_for_test')
current_data = load_visitor_data()
user_data = {}
if user_id_str and user_id_str in current_data:
user_data = current_data[user_id_str]
user_data['id'] = user_id_str
else:
user_data = {
"id": "N/A", "bonuses": 0, "debts": 0, "first_name": "Гость",
"history": [], "debt_history": [], "invoices": []
}
if page == 'card':
org_info = load_org_info()
return render_template_string(CARD_TEMPLATE, user=user_data, org_info=org_info, clean_phone=clean_phone_number)
elif page == 'history':
bonus_history = [dict(item, transaction_type='bonus') for item in user_data.get('history', [])]
debt_history = [dict(item, transaction_type='debt') for item in user_data.get('debt_history', [])]
user_data['combined_history'] = sorted(bonus_history + debt_history, key=lambda x: x['date'], reverse=True)
return render_template_string(HISTORY_TEMPLATE, user=user_data)
elif page == 'invoices':
if 'invoices' in user_data and user_data['invoices']:
user_data['invoices'] = sorted(user_data['invoices'], key=lambda x: x['date'], reverse=True)
return render_template_string(INVOICES_TEMPLATE, user=user_data)
else: # home
return render_template_string(MAIN_TEMPLATE, user=user_data)
@app.route('/verify', methods=['POST'])
def verify_data():
try:
req_data = request.get_json()
init_data_str = req_data.get('initData')
if not init_data_str:
return jsonify({"status": "error", "message": "Missing initData"}), 400
user_data_parsed, is_valid = verify_telegram_data(init_data_str)
user_info_dict = json.loads(unquote(user_data_parsed['user'][0])) if user_data_parsed and 'user' in user_data_parsed else {}
if is_valid and user_info_dict.get('id'):
tg_user_id = str(user_info_dict['id'])
now = datetime.now(BISHKEK_TZ)
all_data = load_visitor_data()
existing_user_key = next((k for k, v in all_data.items() if str(v.get('telegram_id')) == tg_user_id), None)
if existing_user_key:
user_entry = all_data[existing_user_key]
user_entry.update({
'first_name': user_info_dict.get('first_name'), 'last_name': user_info_dict.get('last_name'),
'username': user_info_dict.get('username'), 'photo_url': user_info_dict.get('photo_url'),
'visited_at': now.timestamp(), 'visited_at_str': now.strftime('%Y-%m-%d %H:%M:%S')
})
user_id_to_save = existing_user_key
else:
new_user_id = generate_unique_id(all_data)
user_entry = {
'id': new_user_id, 'telegram_id': tg_user_id, 'bonuses': 0, 'history': [], 'debts': 0, 'debt_history': [], 'invoices': [],
'first_name': user_info_dict.get('first_name'), 'last_name': user_info_dict.get('last_name'),
'username': user_info_dict.get('username'), 'photo_url': user_info_dict.get('photo_url'),
'phone_number': None, 'visited_at': now.timestamp(), 'visited_at_str': now.strftime('%Y-%m-%d %H:%M:%S')
}
user_id_to_save = new_user_id
save_visitor_data({user_id_to_save: user_entry})
return jsonify({"status": "ok", "verified": True, "user_id": user_id_to_save})
else:
return jsonify({"status": "error", "verified": is_valid, "message": "Invalid data or missing user ID"}), 403
except Exception as e:
return jsonify({"status": "error", "message": "Internal server error"}), 500
@app.route('/admin')
def admin_panel():
current_data = load_visitor_data()
org_info = load_org_info()
users_list = [dict(v, id=k) for k, v in current_data.items()]
summary_stats = {
"total_users": len(users_list),
"total_bonuses": sum(u.get('bonuses', 0) for u in users_list),
"total_debts": sum(u.get('debts', 0) for u in users_list),
"users_with_debt": sum(1 for u in users_list if u.get('debts', 0) > 0)
}
return render_template_string(ADMIN_TEMPLATE, users=users_list, summary=summary_stats, org_info=org_info)
@app.route('/admin/update_organization_info', methods=['POST'])
def update_organization_info():
try:
data = request.form
phones = [p.strip() for p in data.get('phones', '').splitlines() if p.strip()]
links_raw = [p.strip() for p in data.get('links', '').splitlines() if p.strip()]
links = []
for line in links_raw:
parts = [p.strip() for p in line.split(',', 1)]
if len(parts) == 2:
links.append({"label": parts[0], "url": parts[1]})
org_info = {
"name": data.get('name', ''),
"phones": phones,
"address": data.get('address', ''),
"links": links
}
save_org_info(org_info)
return redirect(url_for('admin_panel'))
except Exception:
return "Error", 500
@app.route('/admin/add_client', methods=['POST'])
def add_client():
try:
data = request.get_json()
phone_number = data.get('phone_number')
first_name = data.get('first_name')
if not phone_number or not first_name:
return jsonify({"status": "error", "message": "Имя и номер телефона обязательны."}), 400
all_data = load_visitor_data()
if any(u.get('phone_number') == phone_number for u in all_data.values()):
return jsonify({"status": "error", "message": "Клиент с таким номером уже существует."}), 409
now = datetime.now(BISHKEK_TZ)
new_id = generate_unique_id(all_data)
new_client = {
'id': new_id, 'telegram_id': None, 'first_name': first_name, 'phone_number': phone_number,
'bonuses': 0, 'history': [], 'debts': 0, 'debt_history': [], 'invoices': [],
'visited_at': now.timestamp(), 'visited_at_str': now.strftime('%Y-%m-%d %H:%M:%S'),
'last_name': None, 'username': None, 'photo_url': None
}
save_visitor_data({new_id: new_client})
return jsonify({"status": "ok", "message": "Клиент успешно добавлен!"}), 201
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
@app.route('/admin/add_transaction', methods=['POST'])
def add_transaction():
try:
data = request.get_json()
user_id, purchase, deduct = str(data.get('user_id')), float(data.get('purchase_amount', 0)), float(data.get('deduct_amount', 0))
add_debt, repay_debt = float(data.get('add_debt_amount', 0)), float(data.get('repay_debt_amount', 0))
all_data = load_visitor_data()
if user_id not in all_data: return jsonify({"status": "error", "message": "User not found"}), 404
user = all_data[user_id]
now = datetime.now(BISHKEK_TZ)
accrual_amount = purchase * 0.02
new_balance = user.get('bonuses', 0) + accrual_amount
if deduct > new_balance: return jsonify({"status": "error", "message": "Недостаточно бонусов для списания"}), 400
user['bonuses'] = new_balance - deduct
if accrual_amount > 0: user.setdefault('history', []).append({"type": "accrual", "amount": accrual_amount, "description": f"Начисление с покупки {purchase}", "date": now.isoformat(), "date_str": now.strftime('%Y-%m-%d %H:%M:%S')})
if deduct > 0: user.setdefault('history', []).append({"type": "deduction", "amount": deduct, "description": "Списание бонусов", "date": now.isoformat(), "date_str": now.strftime('%Y-%m-%d %H:%M:%S')})
if repay_debt > user.get('debts', 0): return jsonify({"status": "error", "message": "Сумма погашения превышает долг"}), 400
user['debts'] = user.get('debts', 0) + add_debt - repay_debt
if add_debt > 0: user.setdefault('debt_history', []).append({"type": "accrual", "amount": add_debt, "description": "Добавление долга", "date": now.isoformat(), "date_str": now.strftime('%Y-%m-%d %H:%M:%S')})
if repay_debt > 0: user.setdefault('debt_history', []).append({"type": "payment", "amount": repay_debt, "description": "Погашение долга", "date": now.isoformat(), "date_str": now.strftime('%Y-%m-%d %H:%M:%S')})
save_visitor_data({user_id: user})
return jsonify({"status": "ok", "message": "Операция успешна"}), 200
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
@app.route('/admin/add_invoice', methods=['POST'])
def add_invoice():
try:
data = request.get_json()
user_id = str(data.get('user_id'))
items = data.get('items', [])
if not user_id or not items:
return jsonify({"status": "error", "message": "Требуется ID пользователя и товары"}), 400
all_data = load_visitor_data()
if user_id not in all_data:
return jsonify({"status": "error", "message": "User not found"}), 404
user = all_data[user_id]
now = datetime.now(BISHKEK_TZ)
invoice_id = f"{int(now.timestamp()) % 100000}{random.randint(10,99)}"
total_amount = sum(float(item.get('quantity', 0)) * float(item.get('price_per_unit', 0)) for item in items)
new_invoice = {
"id": invoice_id,
"date": now.isoformat(),
"date_str": now.strftime('%Y-%m-%d %H:%M:%S'),
"items": items,
"total_amount": total_amount
}
user.setdefault('invoices', []).append(new_invoice)
save_visitor_data({user_id: user})
return jsonify({"status": "ok", "message": "Накладная успешно сохранена!"}), 201
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
@app.route('/admin/delete_client', methods=['POST'])
def delete_client():
try:
user_id = str(request.get_json().get('user_id'))
if not user_id: return jsonify({"status": "error", "message": "User ID is required"}), 400
load_visitor_data()
with _data_lock:
if user_id not in visitor_data_cache: return jsonify({"status": "error", "message": "User not found"}), 404
if visitor_data_cache[user_id].get('telegram_id') is not None: return jsonify({"status": "error", "message": "Cannot delete a Telegram-linked user"}), 403
del visitor_data_cache[user_id]
with open(DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(visitor_data_cache, f, ensure_ascii=False, indent=4)
upload_data_to_hf_async()
return jsonify({"status": "ok", "message": "Клиент удален"}), 200
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 500
if __name__ == '__main__':
if HF_TOKEN_READ:
download_files_from_hf()
load_visitor_data()
load_org_info()
if HF_TOKEN_WRITE:
backup_thread = threading.Thread(target=periodic_backup, daemon=True)
backup_thread.start()
app.run(host=HOST, port=PORT, debug=False)