Please provide your UrNetwork credentials to configure the dashboard.
"""
LOGIN_TEMPLATE = """
Sign in to your dashboard
"""
PUBLIC_DASHBOARD_TEMPLATE = """
Monitoring for updates...
Total Paid Data
{{ "%.3f"|format(latest.paid_gb if latest else 0) }} GB
Total Unpaid Data
{{ "%.3f"|format(latest.unpaid_gb if latest else 0) }} GB
Earnings (Last 30 Days)
${{ "%.2f"|format(monthly_earnings) }}
Last Update
{{ latest.timestamp.strftime('%H:%M') if latest else 'N/A' }}
Total Data Provided Over Time (GB)
{% if chart_data.labels %}
{% else %}
Not enough data to display a chart. Check back later.
{% endif %}
Provider stats refresh in: 02:00
Provider Locations
Providers by Country
Loading country data...
"""
PRIVATE_DASHBOARD_TEMPLATE = """
Monitoring for updates...
Total Paid Data
{{ "%.3f"|format(latest.paid_gb if latest else 0) }} GB
Total Unpaid Data
{{ "%.3f"|format(latest.unpaid_gb if latest else 0) }} GB
Total Earnings
${{ "%.2f"|format(total_earnings) }}
Last Update
{{ latest.timestamp.strftime('%H:%M') if latest else 'N/A' }}
Paid vs. Unpaid Data (GB)
{% if chart_data.labels and chart_data.labels|length > 0 %}
{% else %}
Not enough data for this chart.
{% endif %}
Data Change per Interval (GB)
{% if chart_data.deltas and chart_data.deltas|length > 0 %}
{% else %}
Not enough data for this chart.
{% endif %}
Webhook Management
{% for webhook in webhooks %}
{{ webhook.url }}
{% else %}
No webhooks configured.
{% endfor %}
Detailed History
Timestamp
Paid (GB)
Unpaid (GB)
Change (GB)
{% for row in rows %}
{{ row.ts_str }}
{{ "%.3f"|format(row.e.paid_gb) }}
{{ "%.3f"|format(row.e.unpaid_gb) }}
{% if row.delta_gb is not none %}{{ '%+.3f'|format(row.delta_gb) }}{% else %}N/A{% endif %}
{% else %}
No data available.
{% endfor %}
"""
# --- Template Rendering Helper ---
# Load world map data once at startup
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 for embedding.")
except Exception as e:
logging.error(f"Could not download and cache world map data. Map will be disabled. Error: {e}")
def render_page(template_content, **context):
"""Renders a page by injecting its content into the main layout."""
if context.get('is_public_dashboard'):
context['world_map_data'] = WORLD_MAP_DATA
content = render_template_string(template_content, **context)
return render_template_string(LAYOUT_TEMPLATE, content=content, **context)
# --- Middleware and Before Request Handlers ---
@app.before_request
def check_installation():
"""Before each request, check if the app is installed."""
if request.endpoint not in ['install', 'static', 'get_locations', 'get_transfer_data'] and not is_installed():
return redirect(url_for('install'))
g.now = datetime.datetime.utcnow()
@app.context_processor
def inject_layout_vars():
"""Inject variables into all templates for access."""
return {"now": g.now}
# --- Routes ---
@app.route('/install', methods=['GET', 'POST'])
def install():
if is_installed():
flash("Application is already installed.", "info")
return redirect(url_for('public_dashboard'))
if request.method == 'POST':
user = request.form.get('ur_user')
password = request.form.get('ur_pass')
if not user or not password:
flash("Username and Password are required.", "error")
return render_page(INSTALL_TEMPLATE, title="Setup")
jwt = get_jwt_from_credentials(user, password)
if not jwt:
flash("Could not verify credentials with UrNetwork API. Please check them and try again.", "error")
return render_page(INSTALL_TEMPLATE, title="Setup")
config = {
"UR_USER": user,
"UR_PASS": password,
"SECRET_KEY": os.urandom(24).hex()
}
if save_env_file(config):
with app.app_context():
db.create_all()
session['logged_in'] = True
flash("Installation successful! Your dashboard is now configured.", "info")
return redirect(url_for('private_dashboard'))
else:
flash("A server error occurred while saving the configuration file.", "error")
return render_page(INSTALL_TEMPLATE, 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':
username = request.form.get('username')
password = request.form.get('password')
if username == os.getenv('UR_USER') and password == os.getenv('UR_PASS'):
session['logged_in'] = True
flash('You have been logged in successfully.', 'info')
next_url = request.args.get('next')
return redirect(next_url or url_for('private_dashboard'))
else:
flash('Invalid username or password.', 'error')
return render_page(LOGIN_TEMPLATE, title="Login")
@app.route('/logout')
def logout():
session.pop('logged_in', None)
flash('You have been logged out.', 'info')
return redirect(url_for('public_dashboard'))
@app.route('/')
def public_dashboard():
"""Display the public-facing dashboard with summary stats, map, and a graph."""
entries = Stats.query.order_by(Stats.timestamp.asc()).all()
latest = entries[-1] if entries else None
jwt = get_valid_jwt()
payments = fetch_payment_stats(jwt)
_, monthly_earnings = calculate_earnings(payments)
chart_data = {
"labels": [e.timestamp.strftime('%m-%d %H:%M') for e in entries],
"data": [e.paid_gb + e.unpaid_gb for e in entries]
}
latest_timestamp = latest.timestamp.isoformat() if latest else None
return render_page(PUBLIC_DASHBOARD_TEMPLATE, title="Public Dashboard", latest=latest, monthly_earnings=monthly_earnings, chart_data=chart_data, is_public_dashboard=True, latest_timestamp=latest_timestamp)
@app.route('/api/locations')
def get_locations():
"""API endpoint to proxy provider locations, simplifying client-side fetching."""
data = fetch_provider_locations()
if data:
return jsonify(data)
return jsonify({"error": "Failed to fetch location data"}), 500
@app.route('/api/transfer_data')
def get_transfer_data():
"""API endpoint to get all transfer stats for dynamic updates."""
entries = Stats.query.order_by(Stats.timestamp.asc()).all()
if not entries:
return jsonify({"status": "no_data"})
latest_entry = entries[-1]
latest_timestamp = latest_entry.timestamp.isoformat()
jwt = get_valid_jwt()
payments = fetch_payment_stats(jwt)
total_earnings, monthly_earnings = calculate_earnings(payments)
# Data for Public View
public_stats = {
"paid_gb": "%.3f" % latest_entry.paid_gb,
"unpaid_gb": "%.3f" % latest_entry.unpaid_gb,
"last_update": latest_entry.timestamp.strftime('%H:%M')
}
public_chart_data = {
"labels": [e.timestamp.strftime('%m-%d %H:%M') for e in entries],
"data": [(e.paid_gb + e.unpaid_gb) for e in entries]
}
# Data for Private View
desc_entries = sorted(entries, key=lambda x: x.timestamp, reverse=True)
private_rows = []
for i, e in enumerate(desc_entries):
delta_gb = None
if i < len(desc_entries) - 1:
prev_entry = desc_entries[i+1]
delta_gb = (e.paid_gb + e.unpaid_gb) - (prev_entry.paid_gb + prev_entry.unpaid_gb)
private_rows.append({
"ts_str": e.timestamp.strftime("%Y-%m-%d %H:%M:%S"),
"paid_gb": e.paid_gb,
"unpaid_gb": e.unpaid_gb,
"delta_gb": delta_gb
})
rev_entries = list(reversed(desc_entries))
deltas = [r['delta_gb'] for r in reversed(private_rows) if r['delta_gb'] is not None]
private_chart_data = {
"labels": [e.timestamp.strftime('%m-%d %H:%M') for e in rev_entries],
"paid_gb": [e.paid_gb for e in rev_entries],
"unpaid_gb": [e.unpaid_gb for e in rev_entries],
"deltas": deltas
}
return jsonify({
"status": "ok",
"latest_timestamp": latest_timestamp,
"public_stats": public_stats,
"public_chart_data": public_chart_data,
"monthly_earnings": monthly_earnings,
"total_earnings": total_earnings,
"private_rows": private_rows,
"private_chart_data": private_chart_data
})
@app.route('/dashboard')
@login_required
def private_dashboard():
"""Display the detailed private dashboard for the owner."""
entries = Stats.query.order_by(Stats.timestamp.desc()).all()
latest = entries[0] if entries else None
jwt = get_valid_jwt()
payments = fetch_payment_stats(jwt)
total_earnings, _ = calculate_earnings(payments)
webhooks = Webhook.query.all()
rows = []
for i, e in enumerate(entries):
delta_gb = None
if i < len(entries) - 1:
prev_entry = entries[i+1]
delta_gb = (e.paid_gb + e.unpaid_gb) - (prev_entry.paid_gb + prev_entry.unpaid_gb)
rows.append({
"e": e,
"ts_str": e.timestamp.strftime("%Y-%m-%d %H:%M:%S"),
"delta_gb": delta_gb
})
rev_entries = list(reversed(entries))
deltas = [r['delta_gb'] for r in reversed(rows) if r['delta_gb'] is not None]
chart_data = {
"labels": [e.timestamp.strftime('%m-%d %H:%M') for e in rev_entries],
"paid_gb": [e.paid_gb for e in rev_entries],
"unpaid_gb": [e.unpaid_gb for e in rev_entries],
"deltas": deltas if deltas else []
}
latest_timestamp = entries[0].timestamp.isoformat() if entries else None
return render_page(
PRIVATE_DASHBOARD_TEMPLATE,
title="Owner Dashboard",
rows=rows,
latest=latest,
total_earnings=total_earnings,
chart_data=chart_data,
latest_timestamp=latest_timestamp,
webhooks=webhooks
)
@app.route("/trigger", methods=["POST"])
@login_required
def trigger_fetch():
"""Manually trigger a stats fetch."""
try:
jwt = get_valid_jwt()
if not jwt:
raise RuntimeError("Failed to get a valid API token.")
stats_data = fetch_transfer_stats(jwt)
entry = Stats(
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()
flash("Successfully fetched and saved latest stats.", "info")
except Exception as e:
logging.error(f"Manual fetch failed: {e}")
flash(f"Failed to fetch stats: {e}", "error")
return redirect(url_for('private_dashboard'))
@app.route("/clear", methods=["POST"])
@login_required
def clear_db():
"""Delete all stats records from the database."""
try:
num_rows_deleted = db.session.query(Stats).delete()
db.session.commit()
flash(f"Successfully cleared {num_rows_deleted} records from the database.", "info")
except Exception as e:
logging.error(f"Failed to clear database: {e}")
flash("An error occurred while clearing the database.", "error")
db.session.rollback()
return redirect(url_for('private_dashboard'))
@app.route("/webhooks/add", methods=["POST"])
@login_required
def add_webhook():
"""Add a new webhook."""
url = request.form.get("webhook_url")
if not url:
flash("Webhook URL cannot be empty.", "error")
return redirect(url_for('private_dashboard'))
# Basic URL validation
if not url.startswith("http://") and not url.startswith("https://"):
flash("Invalid webhook URL.", "error")
return redirect(url_for('private_dashboard'))
existing = Webhook.query.filter_by(url=url).first()
if existing:
flash("This webhook URL is already registered.", "error")
else:
new_webhook = Webhook(url=url)
db.session.add(new_webhook)
db.session.commit()
flash("Webhook added successfully.", "info")
return redirect(url_for('private_dashboard'))
@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("Webhook deleted successfully.", "info")
else:
flash("Webhook not found.", "error")
return redirect(url_for('private_dashboard'))
# --- Main Entry Point ---
if __name__ == "__main__":
with app.app_context():
if is_installed():
db.create_all()
scheduler.start()
app.run(host="0.0.0.0", port=90, debug=False)