Add Xquik publishing and harden webhook proxy trust

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
     {

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

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

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__":