#!/usr/bin/env python3 # -*- coding: utf-8 -*- 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 logging import threading from huggingface_hub import HfApi, hf_hub_download from huggingface_hub.utils import RepositoryNotFoundError, HfHubHTTPError # --- Configuration --- BOT_TOKEN = os.getenv("BOT_TOKEN", "7566834146:AAGiG4MaTZZvvbTVsqEJVG5SYK5hUlc_Ewo") # Use environment variable or default HOST = '0.0.0.0' PORT = 7860 DATA_FILE = 'data.json' # File to store visited user data # Hugging Face Hub Configuration REPO_ID = "flpolprojects/teledata" HF_TOKEN = os.getenv("HF_TOKEN") # Write token HF_TOKEN_READ = os.getenv("HF_TOKEN_READ", HF_TOKEN) # Read token (defaults to write token if not set) BACKUP_INTERVAL = 900 # Seconds (15 minutes) app = Flask(__name__) app.secret_key = os.urandom(24) # Needed for flash messages or sessions if used later # Logging Setup logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') # --- Hugging Face Hub Functions --- def download_db_from_hf(): if not HF_TOKEN_READ: logging.warning("HF_TOKEN_READ not set. Skipping download from Hugging Face Hub.") return False try: logging.info(f"Attempting to download {DATA_FILE} from {REPO_ID}...") hf_hub_download( repo_id=REPO_ID, filename=DATA_FILE, repo_type="dataset", token=HF_TOKEN_READ, local_dir=".", local_dir_use_symlinks=False, force_download=True, # Ensure we get the latest version resume_download=False ) logging.info(f"{DATA_FILE} successfully downloaded from Hugging Face Hub.") return True except RepositoryNotFoundError: logging.warning(f"Repository {REPO_ID} not found on Hugging Face Hub. Will use/create local file.") return False except HfHubHTTPError as e: if e.response.status_code == 404: logging.warning(f"{DATA_FILE} not found in repository {REPO_ID}. Will use/create local file.") else: logging.error(f"HTTP error downloading {DATA_FILE} from Hugging Face Hub: {e}") return False except Exception as e: logging.error(f"Error downloading {DATA_FILE} from Hugging Face Hub: {e}") return False def upload_db_to_hf(): if not HF_TOKEN: logging.warning("HF_TOKEN not set. Skipping upload to Hugging Face Hub.") return False if not os.path.exists(DATA_FILE): logging.warning(f"{DATA_FILE} not found locally. Skipping upload.") return False try: api = HfApi() logging.info(f"Attempting to upload {DATA_FILE} to {REPO_ID}...") api.upload_file( path_or_fileobj=DATA_FILE, path_in_repo=DATA_FILE, repo_id=REPO_ID, repo_type="dataset", token=HF_TOKEN, commit_message=f"Automated user data backup {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" ) logging.info(f"{DATA_FILE} successfully uploaded to Hugging Face Hub.") return True except Exception as e: logging.error(f"Error uploading {DATA_FILE} to Hugging Face Hub: {e}") return False def periodic_backup(): logging.info(f"Starting periodic backup thread. Interval: {BACKUP_INTERVAL} seconds.") while True: time.sleep(BACKUP_INTERVAL) logging.info("Initiating scheduled backup...") upload_db_to_hf() # --- Data Handling --- def load_users(): # Attempt download first download_db_from_hf() if not os.path.exists(DATA_FILE): logging.warning(f"{DATA_FILE} not found. Initializing empty user data.") return {} try: with open(DATA_FILE, 'r', encoding='utf-8') as f: users_data = json.load(f) if not isinstance(users_data, dict): logging.warning(f"{DATA_FILE} does not contain a valid JSON dictionary. Resetting.") return {} logging.info(f"Loaded {len(users_data)} user records from {DATA_FILE}.") return users_data except json.JSONDecodeError: logging.error(f"Error decoding JSON from {DATA_FILE}. Returning empty data.") # Consider backing up the corrupted file here return {} except Exception as e: logging.error(f"Error loading user data from {DATA_FILE}: {e}") return {} def save_users(users_data): try: with open(DATA_FILE, 'w', encoding='utf-8') as f: json.dump(users_data, f, ensure_ascii=False, indent=4) logging.info(f"Saved {len(users_data)} user records to {DATA_FILE}.") # Attempt upload after saving locally upload_db_to_hf() except Exception as e: logging.error(f"Error saving user data to {DATA_FILE}: {e}") # Load initial data on startup visited_users = load_users() # --- Telegram Verification --- 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: logging.warning("Verification failed: No hash found in initData.") return None, False data_check_list = [] for key, value in sorted(parsed_data.items()): # Ensure values are strings before appending data_check_list.append(f"{key}={value[0]}") data_check_string = "\n".join(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: auth_date = int(parsed_data.get('auth_date', [0])[0]) current_time = int(time.time()) # Allow slightly older data, adjust timeout as needed (e.g., 3600 for 1 hour) if current_time - auth_date > 86400: # 24 hours tolerance logging.warning(f"Telegram InitData is older than 24 hours (Auth Date: {auth_date}, Current: {current_time}).") # logging.info("Telegram data verified successfully.") return parsed_data, True else: logging.warning(f"Data verification failed. Calculated: {calculated_hash}, Received: {received_hash}") return parsed_data, False except Exception as e: logging.error(f"Error verifying Telegram data: {e}") return None, False # --- Templates --- TEMPLATE = """ Morshen Group - IT Holding
Связаться
Лидер инноваций 2025 Международный Холдинг

Создаем будущее IT сегодня

Мы — международный IT холдинг, объединяющий передовые технологические компании для создания прорывных решений мирового уровня в сферах AI, квантовых вычислений и разработки ПО.

Обсудить ваш проект

Экосистема Инноваций

В состав холдинга входят специализированные компании, каждая из которых является экспертом в своей области передовых технологий.

Искусственный интеллект Квантовые технологии Стратегические решения

Флагман холдинга. Занимаемся R&D в области AI и квантовых технологий, разрабатываем передовые бизнес-решения, формирующие будущее индустрии.

3+ Страны
3K+ Клиенты
5+ Лет на рынке
Веб-разработка Мобильные приложения Энтерпрайз ПО

Студия разработки полного цикла. Создаем высокотехнологичные веб-сайты, мобильные приложения и кастомное ПО для бизнеса любого масштаба, используя современные стеки и методологии.

10+ Лет опыта
Highload Сложные проекты
Agile Быстрый запуск
На сайт Связаться

Глобальное Присутствие

Наши решения и команды работают в ключевых регионах Центральной Азии:

Узбекистан
Казахстан
Кыргызстан
""" ADMIN_TEMPLATE = """ Admin - Посетители

Панель Администратора - Посетители

ВНИМАНИЕ: Этот раздел не защищен! Добавьте аутентификацию для реального использования.
{% if users %}
{% for user in users|sort(attribute='visited_at', reverse=true) %}
User Avatar
{{ user.first_name or '' }} {{ user.last_name or '' }}
{% if user.username %}
@{{ user.username }}
{% else %}
Нет username
{% endif %}
ID: {{ user.id }}
Язык: {{ user.language_code or 'N/A' }}
Телефон: Недоступен
Визит: {{ user.visited_at_str }}
{% endfor %}
{% else %}

Пока нет данных о посетителях.

{% endif %}
""" # --- Flask Routes --- @app.route('/') def index(): return render_template_string(TEMPLATE) @app.route('/verify', methods=['POST']) def verify_data(): global visited_users try: data = request.get_json() init_data_str = data.get('initData') if not init_data_str: logging.warning("Verification request missing initData.") return jsonify({"status": "error", "message": "Missing initData"}), 400 user_data_parsed, is_valid = verify_telegram_data(init_data_str) user_info_dict = {} if user_data_parsed and 'user' in user_data_parsed: try: # Decode JSON string within the 'user' field user_json_str = unquote(user_data_parsed['user'][0]) user_info_dict = json.loads(user_json_str) except (KeyError, IndexError, json.JSONDecodeError, TypeError) as e: logging.error(f"Could not parse user JSON from initData: {e} - Data: {user_data_parsed.get('user')}") user_info_dict = {} # Ensure it's a dict even on error if is_valid: user_id = user_info_dict.get('id') if user_id: user_id_str = str(user_id) # Use string keys for JSON consistency now = time.time() update_data = { 'id': user_id, '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'), 'language_code': user_info_dict.get('language_code'), 'visited_at': now, 'visited_at_str': datetime.fromtimestamp(now).strftime('%Y-%m-%d %H:%M:%S UTC') # Explicit UTC } # Update the global dictionary and save visited_users[user_id_str] = update_data save_users(visited_users) # Save after modification logging.info(f"User visit recorded/updated for ID: {user_id_str}") return jsonify({"status": "ok", "verified": True, "user": user_info_dict}), 200 else: logging.warning(f"Verification failed for user ID: {user_info_dict.get('id', 'Unknown')}") return jsonify({"status": "error", "verified": False, "message": "Invalid data"}), 403 except Exception as e: logging.exception("Critical error in /verify endpoint") # Log full traceback return jsonify({"status": "error", "message": "Internal server error"}), 500 @app.route('/admin') def admin_panel(): # WARNING: This route is unprotected! Add proper authentication/authorization for production. # Load fresh data for admin view, though 'visited_users' global should be up-to-date current_users = load_users() users_list = list(current_users.values()) logging.info(f"Admin panel accessed. Displaying {len(users_list)} users.") return render_template_string(ADMIN_TEMPLATE, users=users_list) @app.route('/backup', methods=['POST']) def backup_route(): # Manual backup trigger # WARNING: Unprotected route logging.info("Manual backup requested via /backup route.") if upload_db_to_hf(): # Optionally add a success message (e.g., using flash) pass else: # Optionally add an error message pass return redirect(url_for('admin_panel')) # Redirect back to admin @app.route('/download', methods=['GET']) def download_route(): # Manual download trigger # WARNING: Unprotected route global visited_users logging.info("Manual download requested via /download route.") if download_db_from_hf(): visited_users = load_users() # Reload data after download # Optionally add a success message else: # Optionally add an error message pass return redirect(url_for('admin_panel')) # Redirect back to admin # --- Main Execution --- if __name__ == '__main__': # Initial check for HF tokens if not HF_TOKEN: logging.warning("!!! HF_TOKEN environment variable is not set. Uploads to Hugging Face Hub will be disabled.") if not HF_TOKEN_READ: logging.warning("!!! HF_TOKEN_READ environment variable is not set. Downloads from Hugging Face Hub will be disabled (falling back to local file).") # Start the periodic backup thread if HF_TOKEN: # Only start if upload is possible backup_thread = threading.Thread(target=periodic_backup, daemon=True) backup_thread.start() else: logging.warning("Periodic backup thread not started because HF_TOKEN is not set.") logging.warning("--- SECURITY WARNING ---") logging.warning("The /admin, /backup, /download routes are NOT protected by authentication.") logging.warning("Anyone knowing the URL can access visitor data and trigger actions.") logging.warning("Implement proper security (e.g., password protection, IP restriction) before deploying.") logging.warning("------------------------") logging.info(f"Starting Flask server on http://{HOST}:{PORT}") logging.info(f"Ensure this address is accessible and configured in BotFather for your Mini App.") logging.info(f"Using Bot Token ID: {BOT_TOKEN.split(':')[0]}") logging.info(f"User data file: {DATA_FILE}") logging.info(f"Hugging Face Repo: {REPO_ID}") # Use Waitress or Gunicorn for production instead of app.run() # from waitress import serve # serve(app, host=HOST, port=PORT) app.run(host=HOST, port=PORT, debug=False) # debug=False for production