diff --git a/README.md b/README.md index ee3f07f..7410506 100644 --- a/README.md +++ b/README.md @@ -34,6 +34,7 @@ All alerts can be instantly sent to Telegram, Discord, Twitter and/or Email. - Discord Support using [webhooks](https://support.discord.com/hc/de/articles/228383668-Webhooks-verwenden). - Slack Support using [webhooks](https://api.slack.com/messaging/webhooks). - Twitter Support using the [tweepy](https://github.com/tweepy/tweepy) libary. +- Optional X publishing through [Xquik](https://xquik.com). - Email Support using [smtplib](https://docs.python.org/3/library/smtplib.html). - Alert channels can be enabled or disabled in [`config.py`](https://github.com/fabston/TradingView-Webhook-Bot/blob/master/config.py). - Dynamically send alerts to different Telegram and/or Discord channels. @@ -46,6 +47,8 @@ All alerts can be instantly sent to Telegram, Discord, Twitter and/or Email. 1. Activate it `source TradingView-Webhook-Bot/bin/activate && cd TradingView-Webhook-Bot` 1. Install all requirements `pip install -r requirements.txt` 1. Edit and update [`config.py`](https://github.com/fabston/TradingView-Webhook-Bot/blob/master/config.py) + - To publish through Xquik, enable `send_twitter_alerts`, connect an X account in the Xquik dashboard, and set `TWITTER_BACKEND=xquik`, `XQUIK_API_KEY`, and `XQUIK_ACCOUNT`. The account value can be its X username or connected-account ID. API-key authentication is supported; Xquik also offers OAuth-first MCP access. + - Xquik is an independent third-party service. Not affiliated with X Corp. "Twitter" and "X" are trademarks of X Corp. 1. Setup TradingView alerts. An example alert message would be: ```json { diff --git a/config.py b/config.py index d9428a6..12abe85 100644 --- a/config.py +++ b/config.py @@ -1,8 +1,10 @@ -# ----------------------------------------------- # -# Plugin Name : TradingView-Webhook-Bot # -# Author Name : fabston # -# File Name : config.py # -# ----------------------------------------------- # +# ----------------------------------------------- # +# Plugin Name : TradingView-Webhook-Bot # +# Author Name : fabston # +# File Name : config.py # +# ----------------------------------------------- # + +import os # TradingView Example Alert Message: # { @@ -26,12 +28,19 @@ send_slack_alerts = False slack_webhook = "" # Slack Webhook URL (https://api.slack.com/messaging/webhooks) -# Twitter Settings -send_twitter_alerts = False -tw_ckey = "" -tw_csecret = "" -tw_atoken = "" -tw_asecret = "" +# Twitter Settings +send_twitter_alerts = False +twitter_backend = os.getenv("TWITTER_BACKEND", "twitter") +tw_ckey = "" +tw_csecret = "" +tw_atoken = "" +tw_asecret = "" + +# Xquik Settings. Used when TWITTER_BACKEND is set to "xquik". +xquik_api_url = os.getenv("XQUIK_API_URL", "https://xquik.com/api/v1/x/tweets") +xquik_api_key = os.getenv("XQUIK_API_KEY", "") +xquik_account = os.getenv("XQUIK_ACCOUNT", "") +xquik_timeout = 30 # Email Settings send_email_alerts = False diff --git a/handler.py b/handler.py index 7edabe7..933a6fc 100644 --- a/handler.py +++ b/handler.py @@ -8,16 +8,51 @@ import ssl from email.mime.text import MIMEText -import tweepy -from discord_webhook import DiscordEmbed, DiscordWebhook -from slack_webhook import Slack -from telegram import Bot - -import config - - -def send_alert(data): - msg = data["msg"].encode("latin-1", "backslashreplace").decode("unicode_escape") +import requests +import tweepy +from discord_webhook import DiscordEmbed, DiscordWebhook +from slack_webhook import Slack +from telegram import Bot + +import config + + +def format_twitter_status(msg): + return msg.replace("*", "").replace("_", "").replace("`", "") + + +def send_tweepy_alert(msg): + tw_auth = tweepy.OAuthHandler(config.tw_ckey, config.tw_csecret) + tw_auth.set_access_token(config.tw_atoken, config.tw_asecret) + tw_api = tweepy.API(tw_auth) + tw_api.update_status(status=format_twitter_status(msg)) + + +def send_xquik_alert(msg): + if not config.xquik_api_key or not config.xquik_account: + raise ValueError("XQUIK_API_KEY and XQUIK_ACCOUNT are required") + + response = requests.post( + config.xquik_api_url, + headers={"x-api-key": config.xquik_api_key}, + json={"account": config.xquik_account, "text": format_twitter_status(msg)}, + timeout=config.xquik_timeout, + ) + response.raise_for_status() + return response.json() + + +def send_twitter_alert(msg): + backend = config.twitter_backend.strip().lower() + if backend == "xquik": + return send_xquik_alert(msg) + if backend == "twitter": + return send_tweepy_alert(msg) + raise ValueError("twitter_backend must be 'twitter' or 'xquik'") + + +def send_alert(data): + msg = data["msg"].encode("latin-1", "backslashreplace").decode("unicode_escape") if config.send_telegram_alerts: tg_bot = Bot(token=config.tg_token) try: @@ -65,22 +100,15 @@ def send_alert(data): except Exception as e: print("[X] Slack Error:\n>", e) - if config.send_twitter_alerts: - tw_auth = tweepy.OAuthHandler(config.tw_ckey, config.tw_csecret) - tw_auth.set_access_token(config.tw_atoken, config.tw_asecret) - tw_api = tweepy.API(tw_auth) - try: - tw_api.update_status( - status=msg.replace("*", "").replace("_", "").replace("`", "") - ) - except Exception as e: - print("[X] Twitter Error:\n>", e) + if config.send_twitter_alerts: + try: + send_twitter_alert(msg) + except Exception as e: + print("[X] Twitter Error:\n>", e) - if config.send_email_alerts: - try: - email_msg = MIMEText( - msg.replace("*", "").replace("_", "").replace("`", "") - ) + if config.send_email_alerts: + try: + email_msg = MIMEText(format_twitter_status(msg)) email_msg["Subject"] = config.email_subject email_msg["From"] = config.email_sender email_msg["To"] = config.email_sender diff --git a/main.py b/main.py index ed899bd..174d8da 100644 --- a/main.py +++ b/main.py @@ -11,33 +11,48 @@ app = Flask(__name__) +TRADINGVIEW_WEBHOOK_IPS = { + "34.212.75.30", + "52.32.178.7", + "52.89.214.238", + "54.218.53.128", +} +TRUSTED_PROXY_IPS = {"127.0.0.1", "::1"} + def get_timestamp(): timestamp = time.strftime("%Y-%m-%d %X") return timestamp +def get_client_ip(): + remote_addr = request.remote_addr + if remote_addr in TRUSTED_PROXY_IPS: + forwarded_for = request.headers.get("X-Forwarded-For", "") + if forwarded_for: + return forwarded_for.split(",", 1)[0].strip() + return remote_addr + + @app.route("/webhook", methods=["POST"]) def webhook(): - whitelisted_ips = ['52.89.214.238', '34.212.75.30', '54.218.53.128', '52.32.178.7'] - client_ip = request.headers.get('X-Forwarded-For', request.remote_addr) - if client_ip not in whitelisted_ips: - return jsonify({'message': 'Unauthorized'}), 401 + if get_client_ip() not in TRADINGVIEW_WEBHOOK_IPS: + return jsonify({"message": "Unauthorized"}), 401 try: if request.method == "POST": data = request.get_json() if data["key"] == config.sec_key: print(get_timestamp(), "Alert Received & Sent!") send_alert(data) - return jsonify({'message': 'Webhook received successfully'}), 200 + return jsonify({"message": "Webhook received successfully"}), 200 else: print("[X]", get_timestamp(), "Alert Received & Refused! (Wrong Key)") - return jsonify({'message': 'Unauthorized'}), 401 + return jsonify({"message": "Unauthorized"}), 401 except Exception as e: print("[X]", get_timestamp(), "Error:\n>", e) - return jsonify({'message': 'Error'}), 400 + return jsonify({"message": "Error"}), 400 if __name__ == "__main__":