Telegram Alert

[mcdougelas]

Is that possible creating alert to telegram group using this code

https://api.telegram.org/bot<YourBotToken>/sendMessage?chat_id=<YourChatID>&text={{strategy.order.alert_message}}

[soranoo]

I don't think so, this project is mainly for creating a TV webhook signal which is not expected to activate URL. You may checkout fabston's TradingView-Webhook-Bot it can convent a webhook signal to various forms.

[mcdougelas]

Im trying to add one more step by reply message to telegram bot, but it giving me more duration 🙈

[soranoo]

Seems it took quite a long time to send the tg msg but it should be fine for notification purposes.

[mcdougelas]

Still, I really hope that you can help me provide a solution on how to put the above code in the URL webhook section, so that we can save a few more seconds. Thank you in advance 🙏

Oh, i get helped by gpt about my request and it give me some code for reference

import requests

telegram_bot_token = "YourTelegramBotToken" telegram_chat_id = "YourTelegramChatID" alert_message = "YourAlertMessage"

url = f"https://api.telegram.org/bot{telegram_bot_token}/sendMessage" params = {"chat_id": telegram_chat_id, "text": alert_message}

response = requests.post(url, params=params)

if response.status_code == 200: print("Pesan berhasil dikirim ke Telegram.") else: print(f"Error {response.status_code}: {response.text}")

Can this code be used too?

[soranoo]

All right, I will add the tg feature to it. Give me a day.

[soranoo]

Here you are. It should fit your needs.

import re
import json
import requests
from requests.adapters import HTTPAdapter, Retry
from datetime import datetime

# =====================================
# SETTING
# =====================================

webhook_urls = [
  r"https://yourwebhook.com",
  #r"https://yourwebhook2.com",
  #r"https://yourwebhook3.com",
]

# vv for logging (fill in the webhook URL if you need)
discord_webhook_url = r""

# 4 = info, ok, warning, error
# 3 = ok, warning, error
# 2 = warning, error
# 1 = error only
# 0 = OFF
log_level = 4

# Telegram Bot Notification (leave blank if you don't need)
tg_bot_token = ""
tg_chat_id = ""

# =====================================
# CODE
# =====================================

show_welcome_msg = False

url_regex = "https?://([A-Za-z_0-9.-]+).*"

dc_embeds_sample = [{
    "type":"rich",
    "title": "",
    "description": "",
    "color": 5533306,
    "footer": {
        "text": "Developed by Freeman(soranoo)",
        "icon_url": "https://avatars.githubusercontent.com/u/46896789?v=4"
      },
    }]

class log:
    def info(payload):
        if (log_level < 4 or log_level < 0):
            return
        embeds = dc_embeds_sample.copy()
        embeds[0]["title"] = "INFO"
        embeds[0]["description"] = payload
        embeds[0]["color"] = 5533306
        send_msg_to_dc("", embeds)

    def ok(payload:str):
        if (log_level < 3 or log_level < 0):
            return
        embeds = dc_embeds_sample.copy()
        embeds[0]["title"] = "OK"
        embeds[0]["description"] = payload
        embeds[0]["color"] = 5763719
        send_msg_to_dc("", embeds)

    def warning(payload:str):
        if (log_level < 2 or log_level < 0):
            return
        embeds = dc_embeds_sample.copy()
        embeds[0]["title"] = "WARNING"
        embeds[0]["description"] = payload
        embeds[0]["color"] = 16776960
        send_msg_to_dc("", embeds)

    def error(payload:str):
        if (log_level < 1 or log_level < 0):
            return
        embeds = dc_embeds_sample.copy()
        embeds[0]["title"] = "ERROR"
        embeds[0]["description"] = payload
        embeds[0]["color"] = 15548997
        send_msg_to_dc("", embeds)

def extract_domain_name(url:str):
    m = re.search(url_regex, url)
    if m:
        return m.group(1)
    return None

def post_request(webhook_url:str, payload:str or json, auto_json_dumps:bool = True):
    headers = {
            "Content-Type": "application/json",
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.11 (KHTML, like Gecko) Chrome/23.0.1271.64 Safari/537.11"
        }
    data = json.dumps(payload) if auto_json_dumps else payload
    # start a session
    session = requests.Session()
    # set retry policy
    retry = Retry(connect=3, backoff_factor=0.5)
    adapter = HTTPAdapter(max_retries=retry)
    # add adapter to session
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    # send request
    response = session.post(webhook_url, data=data, headers=headers)
    return response

def send_msg_to_tg(message):
    if (len(tg_bot_token.strip()) == 0 or len(tg_chat_id.strip()) == 0):
        return # empty token or chat ID
    payload = json.dumps({'chat_id': f"{tg_chat_id}", 'text': f"{message}"})
    post_request(f"https://api.telegram.org/bot{tg_bot_token}/sendMessage", payload, False)

def send_msg_to_dc(message, embeds=None):
    if (len(discord_webhook_url.strip()) == 0):
        return # empty URL
    if extract_domain_name(discord_webhook_url) is None:
        print("Invalid Discord Webhook URL")
        return
    payload = json.dumps({'content': f"{message}", "embeds":embeds} )
    post_request(discord_webhook_url, payload, False)

def send_webhook(payload:str):
    # load json
    try:
        payload = json.loads(payload)
    except ValueError as err:
        pass
    # loop
    for webhook_url in webhook_urls:
        domain = extract_domain_name(webhook_url)
        if domain is None:
            log.warning(f"Send webhook failed.\nReason: invalid URL <{webhook_url}>\nContent: {payload}")
            continue
        currentTime = datetime.now()
        res = post_request(webhook_url, payload)
        time_used = (datetime.now()-currentTime).total_seconds() * 1000
        if (res.status_code == 200):
            log.ok(f"Webhook Sent!\n\nProcess Time: {time_used}ms\nDomain: {domain}\nFull URL: {webhook_url}\nContent: {payload}")
        else:
            log.error(f"Send webhook failed.\nReason: {res.text}\nStatus: {res.status_code}\n\nDomain: {domain}\nFull URL: {webhook_url}\nContent: {payload}")

def handler(pd: "pipedream"):
    global show_welcome_msg
    if not show_welcome_msg:
        log.info("You are GOOD to go!\n\n")
        show_welcome_msg = True

    # extract data from previous steps
    content = pd.steps["trigger"]["event"]["body"]["text"]
    from_email = pd.steps["trigger"]["event"]["headers"]["from"]["value"][0]["address"]

    if from_email in ["noreply@tradingview.com"]:
        send_webhook(content)
        send_msg_to_tg(content)