import os import sys from flask import Flask, render_template_string, jsonify, request import markdown import time import requests import subprocess app = Flask(__name__) # Cache storage: filepath -> (mtime, html_content) _content_cache = {} # Constants for paths to avoid re-calculating on every request BASE_DIR = os.path.dirname(os.path.abspath(__file__)) README_PATH = os.path.join(BASE_DIR, '..', 'README.md') VERIFICATION_PATH = os.path.join(BASE_DIR, '..', 'VERIFICATION.md') # HTML Template DASHBOARD_HTML = """ MQL5 Trading Automation Dashboard
System Status Webhook API Documentation

System Status ONLINE

MQL5 Trading Automation is running.

{{ html_verification|safe }}

Webhook API Status

Endpoint: /api/signal

Method: POST

Status: Active

This endpoint receives signals from MQL5 Expert Advisors and forwards them to Telegram.

Project Documentation

{{ html_readme|safe }}
""" # Global to store compiled template DASHBOARD_TEMPLATE = None def get_cached_markdown(filepath): try: stat_result = os.stat(filepath) except OSError: return None try: mtime = stat_result.st_mtime if filepath in _content_cache: cached_mtime, cached_html = _content_cache[filepath] if cached_mtime == mtime: return cached_html with open(filepath, 'r', encoding='utf-8') as f: content = f.read() html_content = markdown.markdown(content) _content_cache[filepath] = (mtime, html_content) return html_content except Exception as e: print(f"Error reading/converting {filepath}: {e}") return None @app.route('/health') def health_check(): return jsonify({ "status": "healthy", "webhook_active": True, "timestamp": time.time() }) @app.route('/api/signal', methods=['POST']) def handle_signal(): try: webhook_key = os.environ.get('SIGNAL_WEBHOOK_KEY') if webhook_key: provided_key = request.headers.get('X-Api-Key') or request.args.get('key') if provided_key != webhook_key: return jsonify({"status": "error", "message": "Unauthorized"}), 401 data = request.get_json() if not data: return jsonify({"status": "error", "message": "Invalid JSON payload"}), 400 event = data.get('event', 'signal') message = data.get('message', '') if not message: return jsonify({"status": "error", "message": "Missing message field"}), 400 print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Webhook received: {event}") # Forward to Telegram bot_token = os.environ.get('TELEGRAM_BOT_TOKEN') allowed_users_raw = os.environ.get('TELEGRAM_ALLOWED_USER_IDS', '') if bot_token and allowed_users_raw: allowed_users = [u.strip() for u in allowed_users_raw.split(',') if u.strip()] for user_id in allowed_users: try: tel_url = f"https://api.telegram.org/bot{bot_token}/sendMessage" payload = { "chat_id": user_id, "text": f"🔔 MQL5 Signal: {event}\n\n{message}", "parse_mode": "HTML" } resp = requests.post(tel_url, json=payload, timeout=5) if not resp.ok: print(f"Telegram API error for user {user_id}: {resp.text}") except Exception as e: print(f"Failed to send Telegram notification to {user_id}: {e}") return jsonify({ "status": "success", "message": "Signal processed and forwarded", "timestamp": time.time() }) except Exception as e: print(f"Error in webhook handler: {e}") return jsonify({"status": "error", "message": "Internal server error"}), 500 @app.after_request def add_security_headers(response): csp = "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self'" response.headers['Content-Security-Policy'] = csp response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'SAMEORIGIN' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' return response @app.route('/') def dashboard(): global DASHBOARD_TEMPLATE try: html_readme = get_cached_markdown(README_PATH) or "

README.md not found.

" html_verification = get_cached_markdown(VERIFICATION_PATH) or "

VERIFICATION.md not found.

" if DASHBOARD_TEMPLATE is None: DASHBOARD_TEMPLATE = app.jinja_env.from_string(DASHBOARD_HTML) return DASHBOARD_TEMPLATE.render(html_readme=html_readme, html_verification=html_verification, year=2026) except Exception as e: print(f"Dashboard render error: {e}") return "Internal Server Error", 500 def start_bot(): """Start the telegram bot in the background.""" bot_path = os.path.join(BASE_DIR, "telegram_deploy_bot.py") if os.path.exists(bot_path): try: print("Starting Telegram Deployment Bot in the background...") subprocess.Popen([sys.executable, bot_path]) except Exception as e: print(f"Failed to start Telegram bot: {e}") if __name__ == '__main__': # Start the telegram bot in the background start_bot() port = int(os.environ.get('PORT', 8080)) print(f"Starting web dashboard with Webhook support on port {port}...") app.run(host='0.0.0.0', port=port)