# ---------------------------------------- # UrNetwork Stats Dashboard - Enhanced Multi-Account Edition # ---------------------------------------- # Enhanced with modern design, multi-account support, and combined statistics # ---------------------------------------- import os import time import datetime import requests import logging import json from functools import wraps from flask import ( Flask, request, render_template_string, redirect, url_for, flash, session, g, jsonify ) from flask_sqlalchemy import SQLAlchemy from flask_apscheduler import APScheduler from dateutil.parser import isoparse import secrets from string import Template # --- Application Setup & Configuration --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') ENV_FILE = ".env" def load_env(): """Load environment variables from .env file.""" if os.path.exists(ENV_FILE): with open(ENV_FILE, 'r') as f: for line in f: if '=' in line and not line.strip().startswith('#'): key, value = line.strip().split('=', 1) os.environ[key] = value load_env() class Config: """Flask configuration.""" SCHEDULER_API_ENABLED = True # Use absolute path for database BASE_DIR = os.path.abspath(os.path.dirname(__file__)) INSTANCE_DIR = os.path.join(BASE_DIR, 'instance') SQLALCHEMY_DATABASE_URI = os.getenv("DATABASE_URL", f"sqlite:///{os.path.join(INSTANCE_DIR, 'transfer_stats.db')}") SQLALCHEMY_TRACK_MODIFICATIONS = False SECRET_KEY = os.getenv("SECRET_KEY", "default-secret-key-for-initial-setup") UR_API_BASE = "https://api.bringyour.com" FORCE_HTTPS = os.getenv("FORCE_HTTPS", "False").lower() in ('true', '1', 't') ENABLE_ACCOUNT_STATS = os.getenv("ENABLE_ACCOUNT_STATS", "True").lower() in ('true', '1', 't') ENABLE_LEADERBOARD = os.getenv("ENABLE_LEADERBOARD", "True").lower() in ('true', '1', 't') ENABLE_DEVICE_STATS = os.getenv("ENABLE_DEVICE_STATS", "True").lower() in ('true', '1', 't') # --- Flask App Initialization --- # Ensure instance folder exists BEFORE creating the app instance_path = Config.INSTANCE_DIR if not os.path.exists(instance_path): os.makedirs(instance_path) logging.info(f"Created instance directory: {instance_path}") app = Flask(__name__) app.config.from_object(Config) db = SQLAlchemy(app) scheduler = APScheduler() scheduler.init_app(app) # --- Database Models --- class Account(db.Model): """Represents a UrNetwork account.""" __tablename__ = 'accounts' id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(100), unique=True, nullable=False) password = db.Column(db.String(200), nullable=False) nickname = db.Column(db.String(100), nullable=True) is_active = db.Column(db.Boolean, default=True) created_at = db.Column(db.DateTime, server_default=db.func.now()) class Stats(db.Model): """Represents a snapshot of paid vs unpaid bytes at a given timestamp.""" __tablename__ = 'stats' id = db.Column(db.Integer, primary_key=True) account_id = db.Column(db.Integer, db.ForeignKey('accounts.id'), nullable=False) timestamp = db.Column(db.DateTime, server_default=db.func.now()) paid_bytes = db.Column(db.BigInteger, nullable=False) paid_gb = db.Column(db.Float, nullable=False) unpaid_bytes = db.Column(db.BigInteger, nullable=False) unpaid_gb = db.Column(db.Float, nullable=False) account = db.relationship('Account', backref='stats') class Webhook(db.Model): """Represents a webhook URL with an optional custom payload.""" __tablename__ = 'webhook' id = db.Column(db.Integer, primary_key=True) url = db.Column(db.String, unique=True, nullable=False) payload = db.Column(db.Text, nullable=True) class Setting(db.Model): """Represents a key-value setting for the application.""" __tablename__ = 'settings' key = db.Column(db.String(50), primary_key=True) value = db.Column(db.String(100), nullable=False) # --- Helper Functions --- def get_setting(key, default=None): """Gets a setting value from the database.""" setting = Setting.query.get(key) return setting.value if setting else default def get_boolean_setting(key): """Gets a boolean setting from the database.""" value = get_setting(key, 'False') return value.lower() in ('true', '1', 't') def is_installed(): """Check if the application has been configured.""" return os.getenv("SECRET_KEY") and os.getenv("SECRET_KEY") != "default-secret-key-for-initial-setup" def save_env_file(config_data): """Save configuration data to the .env file.""" try: existing_env = {} if os.path.exists(ENV_FILE): with open(ENV_FILE, 'r') as f: for line in f: if '=' in line and not line.strip().startswith('#'): key, value = line.strip().split('=', 1) existing_env[key] = value existing_env.update(config_data) with open(ENV_FILE, "w") as f: for key, value in existing_env.items(): f.write(f"{key}={value}\n") if "ENABLE_ACCOUNT_STATS" not in existing_env: f.write("\n# Feature Flags\n") f.write("ENABLE_ACCOUNT_STATS=True\n") f.write("ENABLE_LEADERBOARD=True\n") f.write("ENABLE_DEVICE_STATS=True\n") if "FORCE_HTTPS" not in existing_env: f.write("\n# Security Settings\n") f.write("FORCE_HTTPS=False\n") load_env() return True except IOError as e: logging.error(f"Failed to write to .env file: {e}") return False def request_with_retry(method, url, retries=3, backoff=5, timeout=30, **kwargs): """Issue an HTTP request with retries.""" last_exc = None for attempt in range(1, retries + 1): try: resp = requests.request(method, url, timeout=timeout, **kwargs) resp.raise_for_status() return resp except requests.exceptions.RequestException as e: last_exc = e logging.warning(f"[{method.upper()} {url}] attempt {attempt}/{retries} failed: {e}") if attempt < retries: time.sleep(backoff) logging.error(f"All {retries} attempts to {method.upper()} {url} failed: {last_exc}") return None def get_jwt_from_credentials(user, password): """Fetch a new JWT token using username and password.""" try: resp = request_with_retry( "post", f"{app.config['UR_API_BASE']}/auth/login-with-password", headers={"Content-Type": "application/json"}, json={"user_auth": user, "password": password}, ) if not resp: raise RuntimeError("API request failed after multiple retries.") data = resp.json() token = data.get("network", {}).get("by_jwt") if not token: err = data.get("message") or data.get("error") or str(data) raise RuntimeError(f"Login failed: {err}") return token except Exception as e: logging.error(f"Could not get JWT from credentials: {e}") return None def get_valid_jwt(account): """Gets a valid JWT for API calls for a specific account.""" if not account: return None return get_jwt_from_credentials(account.username, account.password) # --- API Fetch Functions --- def fetch_transfer_stats(jwt_token): """Retrieve transfer statistics using the provided JWT.""" if not jwt_token: return None resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/transfer/stats", headers={"Authorization": f"Bearer {jwt_token}"}) if not resp: return None data = resp.json() paid = data.get("paid_bytes_provided", 0) unpaid = data.get("unpaid_bytes_provided", 0) return { "paid_bytes": paid, "paid_gb": paid / 1e9, "unpaid_bytes": unpaid, "unpaid_gb": unpaid / 1e9 } def fetch_payment_stats(jwt_token): """Retrieve account payment statistics using the provided JWT.""" if not jwt_token: return [] resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/account/payments", headers={"Authorization": f"Bearer {jwt_token}"}) if not resp: return [] return resp.json().get("account_payments", []) def fetch_account_details(jwt_token): """Fetches various account details like points and referrals.""" if not jwt_token: return {} headers = {"Authorization": f"Bearer {jwt_token}"} details = {} points_resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/account/points", headers=headers) if points_resp: points_data = points_resp.json().get("network_points", []) details['points'] = sum(p.get('point_value', 0) for p in points_data) referral_resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/account/referral-code", headers=headers) if referral_resp: details['referrals'] = referral_resp.json() ranking_resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/network/ranking", headers=headers) if ranking_resp: details['ranking'] = ranking_resp.json().get('network_ranking', {}) return details def fetch_leaderboard(jwt_token): """Fetches the global leaderboard.""" if not jwt_token: return [] headers = {"Authorization": f"Bearer {jwt_token}"} resp = request_with_retry("post", f"{app.config['UR_API_BASE']}/stats/leaderboard", headers=headers, json={}) if not resp: return [] return resp.json().get("earners", []) def fetch_devices(jwt_token): """Fetches the status of all network clients/devices.""" if not jwt_token: return [] headers = {"Authorization": f"Bearer {jwt_token}"} resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/network/clients", headers=headers) if not resp: return [] devices = resp.json().get("clients", []) provide_mode_map = {-1: "Default", 0: "None", 1: "Network", 2: "Friends & Family", 3: "Public", 4: "Stream"} for device in devices: device['provide_mode_str'] = provide_mode_map.get(device.get('provide_mode'), 'Unknown') return devices def remove_device(jwt_token, client_id): """Removes a device from the network.""" if not jwt_token: return False, "Authentication token not available." headers = {"Authorization": f"Bearer {jwt_token}"} payload = {"client_id": client_id} resp = request_with_retry("post", f"{app.config['UR_API_BASE']}/network/remove-client", headers=headers, json=payload) if resp and resp.status_code == 200: data = resp.json() if data.get("error"): return False, data["error"].get("message", "An unknown error occurred.") return True, "Device removed successfully." elif resp: return False, f"API returned status {resp.status_code}." else: return False, "Failed to communicate with API." def calculate_earnings(payments): """Calculate total and monthly earnings from a list of payments.""" total_earnings = 0 monthly_earnings = 0 now = datetime.datetime.now(datetime.timezone.utc) one_month_ago = now - datetime.timedelta(days=30) if not payments: return 0, 0 for payment in payments: if payment.get("completed"): amount = payment.get("token_amount", 0) total_earnings += amount payment_time_str = payment.get("payment_time") if payment_time_str: try: payment_time = isoparse(payment_time_str) if payment_time > one_month_ago: monthly_earnings += amount except (ValueError, TypeError): logging.warning(f"Could not parse payment_time: {payment_time_str}") return total_earnings, monthly_earnings def fetch_provider_locations(): """Retrieve provider locations from the public API.""" resp = request_with_retry("get", f"{app.config['UR_API_BASE']}/network/provider-locations") return resp.json() if resp else None def send_webhook_notification(stats_data, account_nickname=None): """Sends a notification to all configured webhooks.""" with app.app_context(): webhooks = Webhook.query.all() if not webhooks: return for webhook in webhooks: payload_to_send = None try: if webhook.payload: template = Template(webhook.payload) payload_str = template.safe_substitute( account=account_nickname or "Unknown", paid_gb=f"{stats_data['paid_gb']:.3f}", unpaid_gb=f"{stats_data['unpaid_gb']:.3f}", total_gb=f"{(stats_data['paid_gb'] + stats_data['unpaid_gb']):.3f}", update_time=datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') ) payload_to_send = json.loads(payload_str) else: payload_to_send = { "embeds": [{ "title": f"UrNetwork Stats Update - {account_nickname or 'Account'}", "description": "New data has been synced from the UrNetwork API.", "color": 5814783, "fields": [ {"name": "Account", "value": account_nickname or "Unknown", "inline": False}, {"name": "Total Paid Data", "value": f"{stats_data['paid_gb']:.3f} GB", "inline": True}, {"name": "Total Unpaid Data", "value": f"{stats_data['unpaid_gb']:.3f} GB", "inline": True}, {"name": "Total Data Provided", "value": f"{(stats_data['paid_gb'] + stats_data['unpaid_gb']):.3f} GB", "inline": True}, ], "footer": {"text": f"Update Time: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"} }] } requests.post(webhook.url, json=payload_to_send, timeout=10) logging.info(f"Sent webhook notification to {webhook.url}") except (json.JSONDecodeError, TypeError) as e: logging.error(f"Failed to parse or substitute custom payload for webhook {webhook.url}: {e}") except requests.exceptions.RequestException as e: logging.error(f"Failed to send webhook to {webhook.url}: {e}") # --- Authentication Decorator --- def login_required(f): """Decorator to ensure a user is logged in.""" @wraps(f) def decorated_function(*args, **kwargs): if not session.get('logged_in'): flash(g.t['error_login_required'], "error") next_url = request.url if app.config['FORCE_HTTPS']: next_url = next_url.replace('http://', 'https://', 1) return redirect(url_for('login', next=next_url)) return f(*args, **kwargs) return decorated_function # --- Scheduled Jobs --- @scheduler.task(id="log_stats_job", trigger="cron", minute="0,15,30,45") def log_stats_job(): """Scheduled job to fetch and store stats every 15 minutes for all active accounts.""" with app.app_context(): if not is_installed(): logging.warning("log_stats_job skipped: Application is not installed.") return logging.info("Running scheduled stats fetch for all accounts...") accounts = Account.query.filter_by(is_active=True).all() for account in accounts: try: jwt = get_valid_jwt(account) if not jwt: logging.warning(f"Could not authenticate account {account.username}") continue stats_data = fetch_transfer_stats(jwt) if not stats_data: logging.warning(f"Could not fetch stats for account {account.username}") continue entry = Stats( account_id=account.id, paid_bytes=stats_data["paid_bytes"], paid_gb=stats_data["paid_gb"], unpaid_bytes=stats_data["unpaid_bytes"], unpaid_gb=stats_data["unpaid_gb"] ) db.session.add(entry) db.session.commit() logging.info(f"Logged stats for account {account.nickname or account.username} at {entry.timestamp}") send_webhook_notification(stats_data, account.nickname or account.username) except Exception as e: logging.error(f"Failed to fetch stats for account {account.username}: {e}") @scheduler.task(id="cleanup_old_stats_job", trigger="cron", hour="3", minute="0") def cleanup_old_stats_job(): """Scheduled job to delete stats data older than 7 days, runs daily at 3 AM.""" with app.app_context(): if not is_installed(): logging.warning("cleanup_old_stats_job skipped: Application is not installed.") return logging.info("Running daily stats cleanup job...") try: cutoff_date = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=7) num_rows_deleted = db.session.query(Stats).filter(Stats.timestamp < cutoff_date).delete(synchronize_session=False) db.session.commit() if num_rows_deleted > 0: logging.info(f"Successfully deleted {num_rows_deleted} stats records older than 7 days.") else: logging.info("No old stats records found to delete.") except Exception as e: logging.error(f"Scheduled job 'cleanup_old_stats_job' failed: {e}") db.session.rollback() # --- Internationalization (i18n) --- TEXTS = { 'cs': { 'nav_public': 'Veřejný pohled', 'nav_owner_dashboard': 'Panel vlastníka', 'nav_overview': 'Přehled', 'nav_account': 'Účet & Žebříček', 'nav_devices': 'Zařízení', 'nav_settings': 'Nastavení', 'nav_accounts': 'Správa účtů', 'nav_login': 'Přihlásit se', 'nav_logout': 'Odhlásit se', 'title_public_dashboard': 'Veřejný panel', 'title_owner_dashboard': 'Panel vlastníka', 'title_account_page': 'Účet & Žebříček', 'title_devices_page': 'Správa zařízení', 'title_settings': 'Nastavení', 'title_accounts_page': 'Správa účtů', 'title_login': 'Přihlásit se', 'title_setup': 'Počáteční nastavení', 'card_paid_data': 'Celkem placených dat', 'card_unpaid_data': 'Celkem neplacených dat', 'card_earnings_30d': 'Výdělky (30 dní)', 'card_total_earnings': 'Celkové výdělky', 'card_last_update': 'Poslední aktualizace', 'card_account_points': 'Body účtu', 'card_your_rank': 'Vaše pozice', 'card_total_referrals': 'Celkem doporučení', 'card_active_accounts': 'Aktivní účty', 'card_total_devices': 'Celkem zařízení', 'chart_total_data_gb': 'Celková poskytnutá data (GB)', 'chart_paid_vs_unpaid': 'Placená vs. neplacená data (GB)', 'chart_delta': 'Změna dat za interval (GB)', 'chart_paid': 'Placená (GB)', 'chart_unpaid': 'Neplacená (GB)', 'chart_no_data': 'Nedostatek dat pro zobrazení grafu.', 'history_title': 'Detailní historie', 'history_timestamp': 'Časová značka', 'history_change': 'Změna (GB)', 'btn_fetch_now': 'Načíst nyní', 'btn_clear_data': 'Vymazat všechna data', 'btn_add_account': 'Přidat účet', 'confirm_clear_data': 'Jste si jisti? Tímto smažete všechna historická data statistik.', 'next_fetch': 'Další automatické načtení:', 'login_header': 'Přihlaste se do svého panelu', 'login_username': 'Administrátorské heslo', 'login_password': 'Heslo', 'login_button': 'Přihlásit se', 'error_login_required': 'Pro zobrazení této stránky se musíte přihlásit.', 'error_invalid_credentials': 'Neplatné heslo.', 'flash_logged_in': 'Byli jste úspěšně přihlášeni.', 'flash_logged_out': 'Byli jste odhlášeni.', 'flash_settings_saved': 'Nastavení bylo úspěšně uloženo.', 'flash_fetch_success': 'Nejnovější statistiky byly úspěšně načteny a uloženy.', 'flash_fetch_fail': 'Nepodařilo se načíst statistiky: ', 'flash_clear_success': 'Úspěšně smazáno {count} záznamů z databáze.', 'flash_clear_fail': 'Při mazání databáze došlo k chybě.', 'flash_account_added': 'Účet byl úspěšně přidán.', 'flash_account_removed': 'Účet byl úspěšně odstraněn.', 'flash_account_updated': 'Účet byl úspěšně aktualizován.', 'webhook_title': 'Správa webhooků', 'webhook_desc': 'Přidejte webhooky pro zasílání oznámení.', 'webhook_url_label': 'URL adresa webhooku', 'webhook_payload_label': 'Vlastní JSON Payload (volitelné)', 'webhook_payload_placeholder': 'Např.: {"content": "Účet: ${account}, Data: ${total_gb} GB"}', 'webhook_add_btn': 'Přidat Webhook', 'webhook_current': 'Aktuální Webhooky', 'webhook_delete_btn': 'Smazat', 'webhook_flash_added': 'Webhook úspěšně přidán.', 'webhook_flash_deleted': 'Webhook úspěšně smazán.', 'webhook_flash_invalid_url': 'Je vyžadována platná URL adresa webhooku.', 'webhook_flash_exists': 'Tato URL adresa webhooku je již zaregistrována.', 'accounts_username': 'Uživatelské jméno', 'accounts_nickname': 'Přezdívka', 'accounts_status': 'Stav', 'accounts_actions': 'Akce', 'accounts_active': 'Aktivní', 'accounts_inactive': 'Neaktivní', 'accounts_toggle': 'Přepnout', 'accounts_remove': 'Odebrat', 'accounts_add_title': 'Přidat nový účet', 'accounts_add_username': 'UrNetwork uživatelské jméno', 'accounts_add_password': 'UrNetwork heslo', 'accounts_add_nickname': 'Přezdívka (volitelné)', 'combined_stats': 'Kombinovaná statistika', 'individual_accounts': 'Jednotlivé účty', 'view_combined': 'Zobrazit kombinovanou statistiku', 'view_individual': 'Zobrazit jednotlivé účty', 'map_title': 'Lokace poskytovatelů', 'map_legend_title': 'Počet poskytovatelů', 'map_legend_hover': 'Přejeďte myší přes zemi', 'leaderboard_title': 'Žebříček', 'leaderboard_rank': 'Pořadí', 'leaderboard_name': 'Jméno sítě', 'leaderboard_data': 'Poskytnutá data (MiB)', 'devices_title': 'Stav a správa zařízení', 'devices_status': 'Stav', 'devices_name': 'Jméno zařízení', 'devices_id': 'ID klienta', 'devices_mode': 'Režim poskytování', 'devices_account': 'Účet', 'devices_remove': 'Odebrat', 'devices_online': 'Online', 'devices_offline': 'Offline', 'devices_confirm_remove': 'Opravdu chcete odebrat toto zařízení?', 'flash_device_removed': 'Zařízení úspěšně odebráno.', 'flash_device_remove_fail': 'Nepodařilo se odebrat zařízení: ' }, 'en': { 'nav_public': 'Public View', 'nav_owner_dashboard': 'Owner Dashboard', 'nav_overview': 'Overview', 'nav_account': 'Account & Leaderboard', 'nav_devices': 'Devices', 'nav_settings': 'Settings', 'nav_accounts': 'Account Management', 'nav_login': 'Login', 'nav_logout': 'Logout', 'title_public_dashboard': 'Public Dashboard', 'title_owner_dashboard': 'Owner Dashboard', 'title_account_page': 'Account & Leaderboard', 'title_devices_page': 'Device Management', 'title_settings': 'Settings', 'title_accounts_page': 'Account Management', 'title_login': 'Login', 'title_setup': 'Initial Setup', 'card_paid_data': 'Total Paid Data', 'card_unpaid_data': 'Total Unpaid Data', 'card_earnings_30d': 'Earnings (30 Days)', 'card_total_earnings': 'Total Earnings', 'card_last_update': 'Last Update', 'card_account_points': 'Account Points', 'card_your_rank': 'Your Rank', 'card_total_referrals': 'Total Referrals', 'card_active_accounts': 'Active Accounts', 'card_total_devices': 'Total Devices', 'chart_total_data_gb': 'Total Data Provided (GB)', 'chart_paid_vs_unpaid': 'Paid vs. Unpaid Data (GB)', 'chart_delta': 'Data Change per Interval (GB)', 'chart_paid': 'Paid (GB)', 'chart_unpaid': 'Unpaid (GB)', 'chart_no_data': 'Not enough data to display a chart.', 'history_title': 'Detailed History', 'history_timestamp': 'Timestamp', 'history_change': 'Change (GB)', 'btn_fetch_now': 'Fetch Now', 'btn_clear_data': 'Clear All Data', 'btn_add_account': 'Add Account', 'confirm_clear_data': 'Are you sure? This will delete all historical stats data.', 'next_fetch': 'Next automatic fetch:', 'login_header': 'Sign in to your dashboard', 'login_username': 'Admin Password', 'login_password': 'Password', 'login_button': 'Sign in', 'error_login_required': 'You must be logged in to view this page.', 'error_invalid_credentials': 'Invalid password.', 'flash_logged_in': 'You have been logged in successfully.', 'flash_logged_out': 'You have been logged out.', 'flash_settings_saved': 'Settings saved successfully.', 'flash_fetch_success': 'Successfully fetched and saved latest stats.', 'flash_fetch_fail': 'Failed to fetch stats: ', 'flash_clear_success': 'Successfully cleared {count} records from the database.', 'flash_clear_fail': 'An error occurred while clearing the database.', 'flash_account_added': 'Account added successfully.', 'flash_account_removed': 'Account removed successfully.', 'flash_account_updated': 'Account updated successfully.', 'webhook_title': 'Webhook Management', 'webhook_desc': 'Add webhooks to send notifications.', 'webhook_url_label': 'Webhook URL', 'webhook_payload_label': 'Custom JSON Payload (optional)', 'webhook_payload_placeholder': 'E.g.: {"content": "Account: ${account}, Data: ${total_gb} GB"}', 'webhook_add_btn': 'Add Webhook', 'webhook_current': 'Current Webhooks', 'webhook_delete_btn': 'Delete', 'webhook_flash_added': 'Webhook added successfully.', 'webhook_flash_deleted': 'Webhook deleted successfully.', 'webhook_flash_invalid_url': 'A valid Webhook URL is required.', 'webhook_flash_exists': 'This webhook URL is already registered.', 'accounts_username': 'Username', 'accounts_nickname': 'Nickname', 'accounts_status': 'Status', 'accounts_actions': 'Actions', 'accounts_active': 'Active', 'accounts_inactive': 'Inactive', 'accounts_toggle': 'Toggle', 'accounts_remove': 'Remove', 'accounts_add_title': 'Add New Account', 'accounts_add_username': 'UrNetwork Username', 'accounts_add_password': 'UrNetwork Password', 'accounts_add_nickname': 'Nickname (optional)', 'combined_stats': 'Combined Statistics', 'individual_accounts': 'Individual Accounts', 'view_combined': 'View Combined Stats', 'view_individual': 'View Individual Accounts', 'map_title': 'Provider Locations', 'map_legend_title': 'Provider Count', 'map_legend_hover': 'Hover over a country', 'leaderboard_title': 'Leaderboard', 'leaderboard_rank': 'Rank', 'leaderboard_name': 'Network Name', 'leaderboard_data': 'Data Provided (MiB)', 'devices_title': 'Device Status & Management', 'devices_status': 'Status', 'devices_name': 'Device Name', 'devices_id': 'Client ID', 'devices_mode': 'Provide Mode', 'devices_account': 'Account', 'devices_remove': 'Remove', 'devices_online': 'Online', 'devices_offline': 'Offline', 'devices_confirm_remove': 'Are you sure you want to remove this device?', 'flash_device_removed': 'Device removed successfully.', 'flash_device_remove_fail': 'Failed to remove device: ' } } def get_locale(): """Detects the best language match from request headers.""" return request.accept_languages.best_match(TEXTS.keys()) or 'en' # --- HTML Templates --- LAYOUT_TEMPLATE = """ {{ title }}
{% with messages = get_flashed_messages(with_categories=true) %} {% if messages %} {% for category, message in messages %}
{{ message }}
{% endfor %} {% endif %} {% endwith %} {{ content | safe }}
""" INSTALL_TEMPLATE = """

{{ title }}

Nastavte administrátorské heslo pro váš dashboard.

Poté budete moci přidat své UrNetwork účty v nastavení.

""" LOGIN_TEMPLATE = """

{{ title }}

{{ t.login_header }}

""" PUBLIC_DASHBOARD_TEMPLATE = """

{{ title }}

{{ t.card_paid_data }}
{{ "%.3f"|format(combined.paid_gb) }} GB
{{ t.card_unpaid_data }}
{{ "%.3f"|format(combined.unpaid_gb) }} GB
{{ t.card_active_accounts }}
{{ active_accounts }}
{{ t.card_earnings_30d }}
${{ "%.2f"|format(monthly_earnings) }}

{{ t.chart_total_data_gb }} - {{ t.combined_stats }}

{% if chart_data.labels %}{% else %}

{{ t.chart_no_data }}

{% endif %}
{% for account, acc_data in account_charts.items() %}

{{ t.chart_total_data_gb }}

{% if acc_data.labels %}{% else %}

{{ t.chart_no_data }}

{% endif %}
{% endfor %}

{{ t.map_title }}

""" ACCOUNTS_MANAGE_TEMPLATE = """

{{ title }}

{{ t.accounts_add_title }}

{{ t.individual_accounts }}

{% for account in accounts %} {% else %} {% endfor %}
{{ t.accounts_nickname }} {{ t.accounts_username }} {{ t.accounts_status }} {{ t.accounts_actions }}
{{ account.username }} {{ t.accounts_active if account.is_active else t.accounts_inactive }}
Zatím nemáte přidané žádné účty.
""" PRIVATE_DASHBOARD_REACT_TEMPLATE = """
""" SETTINGS_TEMPLATE = """

{{ title }}

{{ t.webhook_title }}

{{ t.webhook_desc }}

{{ t.webhook_current }}

{% for webhook in webhooks %}

{{ webhook.url }}

{{ webhook.payload or 'Default Discord Payload' }}
{% else %}

Zatím nemáte nakonfigurované žádné webhooky.

{% endfor %}
""" # World map data WORLD_MAP_DATA = None try: map_url = "https://d2ad6b4ur7yvpq.cloudfront.net/naturalearth-3.3.0/ne_110m_admin_0_countries.geojson" response = requests.get(map_url, timeout=15) response.raise_for_status() WORLD_MAP_DATA = response.json() logging.info("Successfully loaded world map GeoJSON data.") except Exception as e: logging.error(f"Could not download world map data: {e}") def render_page(template_content, **context): """Renders a page by injecting its content into the main layout.""" context['world_map_data'] = WORLD_MAP_DATA content = render_template_string(template_content, t=g.t, **context) return render_template_string(LAYOUT_TEMPLATE, content=content, t=g.t, **context) # --- Middleware and Before Request Handlers --- @app.before_request def force_secure(): """Redirect non-secure requests to HTTPS if FORCE_HTTPS is enabled.""" if app.config['FORCE_HTTPS'] and not request.is_secure: if request.headers.get('X-Forwarded-Proto', '').lower() != 'https': secure_url = request.url.replace('http://', 'https://', 1) return redirect(secure_url, code=301) @app.before_request def check_installation_and_init(): """Before each request, check if app is installed and set up globals.""" g.locale = get_locale() g.t = TEXTS[g.locale] if request.endpoint not in ['install', 'static'] and not is_installed(): return redirect(url_for('install')) if is_installed() and request.endpoint not in ['static']: with app.app_context(): if not Setting.query.first(): logging.info("First run: Initializing settings.") settings_to_add = [ Setting(key='ENABLE_ACCOUNT_STATS', value=str(app.config['ENABLE_ACCOUNT_STATS'])), Setting(key='ENABLE_LEADERBOARD', value=str(app.config['ENABLE_LEADERBOARD'])), Setting(key='ENABLE_DEVICE_STATS', value=str(app.config['ENABLE_DEVICE_STATS'])), ] db.session.bulk_save_objects(settings_to_add) db.session.commit() g.now = datetime.datetime.utcnow() @app.context_processor def inject_layout_vars(): """Inject variables into all templates.""" return {"now": g.now, "t": g.t} # --- Routes --- @app.route('/install', methods=['GET', 'POST']) def install(): if is_installed(): flash("Aplikace je již nainstalována.", "info") return redirect(url_for('public_dashboard')) if request.method == 'POST': admin_pass = request.form.get('admin_pass') admin_pass_confirm = request.form.get('admin_pass_confirm') if not admin_pass or not admin_pass_confirm: flash("Vyplňte obě pole s heslem.", "error") return render_page(INSTALL_TEMPLATE, title=g.t['title_setup']) if admin_pass != admin_pass_confirm: flash("Hesla se neshodují.", "error") return render_page(INSTALL_TEMPLATE, title=g.t['title_setup']) config_data = { "SECRET_KEY": secrets.token_hex(24), "ADMIN_PASSWORD": admin_pass } if save_env_file(config_data): with app.app_context(): db.create_all() session['logged_in'] = True flash("Dashboard byl úspěšně nainstalován! Nyní můžete přidat své UrNetwork účty.", "info") return redirect(url_for('accounts_manage')) else: flash("Chyba při ukládání konfigurace.", "error") return render_page(INSTALL_TEMPLATE, title=g.t['title_setup']) @app.route('/login', methods=['GET', 'POST']) def login(): if session.get('logged_in'): return redirect(url_for('private_dashboard')) if request.method == 'POST': password = request.form.get('password') admin_password = os.getenv('ADMIN_PASSWORD') if password and password == admin_password: session['logged_in'] = True flash(g.t['flash_logged_in'], 'info') next_url = request.args.get('next') return redirect(next_url or url_for('private_dashboard')) else: flash(g.t['error_invalid_credentials'], 'error') return render_page(LOGIN_TEMPLATE, title=g.t['title_login']) @app.route('/logout') def logout(): session.pop('logged_in', None) flash(g.t['flash_logged_out'], 'info') return redirect(url_for('public_dashboard')) @app.route('/') def public_dashboard(): """Display the public-facing dashboard with combined stats and individual account charts.""" accounts = Account.query.filter_by(is_active=True).all() active_accounts = len(accounts) # Get combined statistics combined_paid = 0 combined_unpaid = 0 monthly_earnings_total = 0 account_charts = {} colors = ['#667eea', '#764ba2', '#f59e0b', '#10b981', '#ef4444', '#3b82f6'] for idx, account in enumerate(accounts): latest_stat = Stats.query.filter_by(account_id=account.id).order_by(Stats.timestamp.desc()).first() if latest_stat: combined_paid += latest_stat.paid_gb combined_unpaid += latest_stat.unpaid_gb # Get JWT and payments jwt = get_valid_jwt(account) if jwt: payments = fetch_payment_stats(jwt) _, monthly = calculate_earnings(payments) monthly_earnings_total += monthly # Get chart data for this account entries = Stats.query.filter_by(account_id=account.id).order_by(Stats.timestamp.asc()).all() if entries: account_name = account.nickname or account.username color = colors[idx % len(colors)] account_charts[account_name] = { "labels": [e.timestamp.strftime('%m-%d %H:%M') for e in entries], "data": [e.paid_gb + e.unpaid_gb for e in entries], "color": color, "bgColor": f"rgba({int(color[1:3], 16)}, {int(color[3:5], 16)}, {int(color[5:7], 16)}, 0.2)" } combined = { "paid_gb": combined_paid, "unpaid_gb": combined_unpaid } # Combined chart data all_entries = Stats.query.order_by(Stats.timestamp.asc()).all() time_grouped = {} for entry in all_entries: time_key = entry.timestamp.strftime('%m-%d %H:%M') if time_key not in time_grouped: time_grouped[time_key] = {"paid": 0, "unpaid": 0} time_grouped[time_key]["paid"] += entry.paid_gb time_grouped[time_key]["unpaid"] += entry.unpaid_gb chart_data = { "labels": list(time_grouped.keys()), "data": [time_grouped[k]["paid"] + time_grouped[k]["unpaid"] for k in time_grouped.keys()] } return render_page( PUBLIC_DASHBOARD_TEMPLATE, title=g.t['title_public_dashboard'], combined=combined, active_accounts=active_accounts, monthly_earnings=monthly_earnings_total, chart_data=chart_data, account_charts=account_charts ) @app.route('/dashboard') @login_required def private_dashboard(): react_props = get_react_props('overview') return render_page(PRIVATE_DASHBOARD_REACT_TEMPLATE, title=g.t['title_owner_dashboard'], react_props=react_props) def get_react_props(initial_page): """Helper to build the props dictionary for the React app.""" return { "initial_page": initial_page, "translations": g.t } # --- Account Management Routes --- @app.route('/accounts') @login_required def accounts_manage(): accounts = Account.query.all() return render_page(ACCOUNTS_MANAGE_TEMPLATE, title=g.t['title_accounts_page'], accounts=accounts) @app.route('/accounts/add', methods=['POST']) @login_required def accounts_add(): username = request.form.get('username') password = request.form.get('password') nickname = request.form.get('nickname') if not username or not password: flash("Uživatelské jméno a heslo jsou povinné.", "error") return redirect(url_for('accounts_manage')) # Verify credentials jwt = get_jwt_from_credentials(username, password) if not jwt: flash("Nepodařilo se ověřit přihlašovací údaje. Zkontrolujte je a zkuste to znovu.", "error") return redirect(url_for('accounts_manage')) # Check if account already exists existing = Account.query.filter_by(username=username).first() if existing: flash("Tento účet je již přidán.", "error") return redirect(url_for('accounts_manage')) account = Account(username=username, password=password, nickname=nickname) db.session.add(account) db.session.commit() flash(g.t['flash_account_added'], "info") return redirect(url_for('accounts_manage')) @app.route('/accounts/toggle/', methods=['POST']) @login_required def accounts_toggle(account_id): account = Account.query.get_or_404(account_id) account.is_active = not account.is_active db.session.commit() flash(g.t['flash_account_updated'], "info") return redirect(url_for('accounts_manage')) @app.route('/accounts/remove/', methods=['POST']) @login_required def accounts_remove(account_id): account = Account.query.get_or_404(account_id) # Delete all stats for this account Stats.query.filter_by(account_id=account_id).delete() db.session.delete(account) db.session.commit() flash(g.t['flash_account_removed'], "info") return redirect(url_for('accounts_manage')) # --- API Routes for React App --- @app.route('/api/dashboard/overview') @login_required def api_overview_data(): accounts = Account.query.filter_by(is_active=True).all() # Combined statistics combined_paid = 0 combined_unpaid = 0 total_earnings = 0 account_charts = {} colors = ['#667eea', '#764ba2', '#f59e0b', '#10b981', '#ef4444', '#3b82f6'] for idx, account in enumerate(accounts): # Get latest stat latest_stat = Stats.query.filter_by(account_id=account.id).order_by(Stats.timestamp.desc()).first() if latest_stat: combined_paid += latest_stat.paid_gb combined_unpaid += latest_stat.unpaid_gb # Get earnings jwt = get_valid_jwt(account) if jwt: payments = fetch_payment_stats(jwt) total, _ = calculate_earnings(payments) total_earnings += total # Get chart data entries = Stats.query.filter_by(account_id=account.id).order_by(Stats.timestamp.asc()).all() if entries: account_name = account.nickname or account.username color = colors[idx % len(colors)] account_charts[account_name] = { "labels": [e.timestamp.strftime('%m-%d %H:%M') for e in entries], "data": [e.paid_gb + e.unpaid_gb for e in entries], "color": color, "bgColor": f"rgba({int(color[1:3], 16)}, {int(color[3:5], 16)}, {int(color[5:7], 16)}, 0.2)" } # Combined chart all_entries = Stats.query.order_by(Stats.timestamp.asc()).all() time_grouped = {} for entry in all_entries: time_key = entry.timestamp.strftime('%m-%d %H:%M') if time_key not in time_grouped: time_grouped[time_key] = {"paid": 0, "unpaid": 0} time_grouped[time_key]["paid"] += entry.paid_gb time_grouped[time_key]["unpaid"] += entry.unpaid_gb combined_chart = { "labels": list(time_grouped.keys()), "paid_gb": [time_grouped[k]["paid"] for k in time_grouped.keys()], "unpaid_gb": [time_grouped[k]["unpaid"] for k in time_grouped.keys()] } return jsonify({ "combined": { "paid_gb": combined_paid, "unpaid_gb": combined_unpaid }, "active_accounts": len(accounts), "total_earnings": total_earnings, "combined_chart": combined_chart, "account_charts": account_charts }) @app.route('/api/dashboard/account') @login_required def api_account_data(): account_id = request.args.get('account_id', 'all') accounts = Account.query.filter_by(is_active=True).all() account_details = {} leaderboard = [] if account_id != 'all': account = Account.query.get(int(account_id)) if account: jwt = get_valid_jwt(account) if jwt: account_details = fetch_account_details(jwt) leaderboard = fetch_leaderboard(jwt) return jsonify({ "accounts": [{"id": a.id, "username": a.username, "nickname": a.nickname} for a in accounts], "account_details": account_details, "leaderboard": leaderboard }) @app.route('/api/dashboard/devices') @login_required def api_devices_data(): accounts = Account.query.filter_by(is_active=True).all() all_devices = [] for account in accounts: jwt = get_valid_jwt(account) if jwt: devices = fetch_devices(jwt) for device in devices: device['account_id'] = account.id device['account_nickname'] = account.nickname or account.username all_devices.extend(devices) return jsonify({"devices": all_devices}) @app.route('/api/dashboard/devices/remove//', methods=["POST"]) @login_required def api_remove_device(account_id, client_id): account = Account.query.get(account_id) if not account: return jsonify({"success": False, "message": "Account not found"}), 404 jwt = get_valid_jwt(account) success, message = remove_device(jwt, client_id) if success: return jsonify({"success": True, "message": g.t['flash_device_removed']}) else: return jsonify({"success": False, "message": f"{g.t['flash_device_remove_fail']}{message}"}), 500 # --- Other Routes --- @app.route('/api/locations') def get_locations(): """API endpoint to proxy provider locations.""" data = fetch_provider_locations() if data: return jsonify(data) return jsonify({"error": "Failed to fetch location data"}), 500 @app.route('/settings', methods=['GET', 'POST']) @login_required def settings(): webhooks = Webhook.query.all() return render_page(SETTINGS_TEMPLATE, title=g.t['title_settings'], webhooks=webhooks) @app.route("/webhooks/add", methods=["POST"]) @login_required def add_webhook(): """Add a new webhook.""" url = request.form.get("webhook_url") payload = request.form.get("payload") if not url or not (url.startswith("http://") or url.startswith("https://")): flash(g.t['webhook_flash_invalid_url'], "error") elif Webhook.query.filter_by(url=url).first(): flash(g.t['webhook_flash_exists'], "error") else: new_webhook = Webhook(url=url, payload=payload if payload.strip() else None) db.session.add(new_webhook) db.session.commit() flash(g.t['webhook_flash_added'], "info") return redirect(url_for('settings')) @app.route("/webhooks/delete/", methods=["POST"]) @login_required def delete_webhook(webhook_id): """Delete a webhook.""" webhook = Webhook.query.get(webhook_id) if webhook: db.session.delete(webhook) db.session.commit() flash(g.t['webhook_flash_deleted'], "info") return redirect(url_for('settings')) # --- Main Entry Point --- if __name__ == "__main__": with app.app_context(): db.create_all() scheduler.start() app.run(host="0.0.0.0", port=90, debug=False)