Skip to content

Instantly share code, notes, and snippets.

@MatteoRagni
Last active June 9, 2016 10:08
Show Gist options
  • Save MatteoRagni/b01292ee4b4937a3cbc11322ab1fa190 to your computer and use it in GitHub Desktop.
Save MatteoRagni/b01292ee4b4937a3cbc11322ab1fa190 to your computer and use it in GitHub Desktop.
A simple telegram notifier to send a message via bot when a simulation is completed, without the need to load an entire library for telegrams bot.
#!/usr/bin/env ruby
require 'net/http'
require 'yaml'
class String
def chunk(l)
raise ArgumentError, "argument must be an integer" unless l.is_a? Fixnum
self.scan /.{1,#{l}}/m
end
end
module Notify
Config = YAML.load_file "/etc/notify-telegram.conf"
Max_length = 4000
def self.send(m, md=false)
raise ArgumentError, "message must be a string" unless m.is_a? String
uri = URI(Notify::Config['uri'] + Notify::Config['token'] + '/sendMessage')
message = {
'chat_id' => Notify::Config['chatid'],
'text' => m,
}
message['parse_mode'] = 'Markdown' if md
res = Net::HTTP.post_form(uri, message)
Notify::log(res)
end
def self.post(m, md=false)
raise ArgumentError, "message must be a string" unless m.is_a? String
if m.size > Notify::Max_length
Notify::log_other("Message too long. Markdown disabled.") if md
m.chunk(Notify::Max_length).each do |k|
Notify::send(k)
end
else
Notify::send(m, md)
end
Notify::send("from: `" + Notify::Config['notifier'] + "`", true)
end
def self.log(res)
if Notify::Config['log']
if res.body =~ /\"ok\":\s*true/m
logs = "#{Time.now()}: [OPERATION] success"
else
logs = "#{Time.now()}: [OPERATION] failed\n"
res.body.chunk(50).each do |l|
logs += "#{" " * (Time.now().to_s.size + 2)} #{l}\n"
end
end
File.open(Notify::Config['log'], 'a') do |f|; f.puts logs; end
end
end
def self.log_other(m)
raise ArgumentError, "m must be a string" unless m.is_a? String
if Notify::Config['log']
logs = "#{Time.now()}: [INTERNAL] #{m}"
File.open(Notify::Config['log'], 'a') do |f|; f.puts logs; end
end
end
end
if __FILE__ == $0 then
message = ARGF.read.chomp
Notify::post(message, true)
end
classdef TelegramNotifier
%TELEGRAMNOTIFIER is an extremely simple class to send a notification
% when a long simulation is ended.
properties
conf_file
config
end
methods
function self = TelegramNotifier(path)
%TELEGRAMNOTIFIER create a new telegram notifier object.
% in: a path to a configuration file (mat) that contains
% 'api_uri', 'api_key', 'chat_id' and 'api_comm'
% * 'api_uri': usually is https://api.telegram.org/bot
% * 'api_comm': usually is /sendMessage
% * 'bot_key': the key provided by TheBotFather
% * 'chat_id': the id of the chat in wich the bot should post the
% notification
% out: a TelegramNotifier object. It is better to create it before
% starting the simulation, and test one notification.
% Check input
assert(ischar(path), ...
'TelegramNotifier:config:bad_argument', ...
'A path to a file is required as char input');
% Loading file
assert(exist(path, 'file') == 2, ...
'TelegramNotifier:config:missing_file', ...
['No file in path ', path]);
% Loading config
self.conf_file = path;
self.config = load(self.conf_file);
% Check config consistency
keys = {'api_uri', 'bot_key', 'chat_id', 'api_comm'};
for i = 1:size(keys,2)
try
self.config.(keys{i});
catch ME
msg = ['Please check ', keys{i} ,' key in your configuration file ', self.conf_file];
cause = MException(['TelegranNotifier:config:', keys{i}], msg);
ME = addCause(ME, cause);
rethrow(ME);
end
end
end
function postMessage(self, mess)
%POSTMESSAGE Post a message of maximum 4090 chars using the bot
% identified by api key provided in configuration file.
% in: mess (a string amx 4090 chars long)
% out: nil
assert(ischar(mess), ...
'TelegramNotifier:postMessage:bad_argument', ...
'A string is required as char input');
assert(size(mess, 2) < 4090, ...
'TelegramNotifier:postMessage:string_too_long', ...
'String maximum length for a post is 4090 (Telegram Api limit)');
uri = [self.config.api_uri, self.config.bot_key, self.config.api_comm];
webwrite(uri, 'chat_id', self.config.chat_id, 'text', mess);
end
end
end
%%EOF
% How to use it:
%
% t = TelegramNotifier('mat-file.mat');
% t.postMessage('Hello World');
% To create the configuration file:
api_uri = 'https://api.telegram.org/api';
bot_key = 'YOUR_BOT_KEY_HERE';
api_comm = '/sendMessage';
chat_id = 'CHAT_ID_NUMBER';
save('mat-file.mat', 'api_uri', 'bot_key', 'api_comm', 'chat_id');
#!/usr/bin/env python3
import requests
from os.path import isfile
import yaml
class string(str):
def chunk(self, length):
assert type(length) is int, "Chunk lenght must be an int. Received a %s" % type(length)
assert length > 0, "Chunk lenght must be positive. Received %d" % length
for i in range(0, len(self), length):
yield self[i:i + length]
class TelegramNotifier:
CONFIG_PATH = '/etc/notify-telegram.conf'
MAX_LENGTH = 3000
def __init__(self):
try:
assert isfile(self.CONFIG_PATH), "Configuration file not found in %s" % self.CONFIG_PATH
with open(self.CONFIG_PATH, 'r') as f:
self.config = yaml.load(f)
self.uri_path = self.config['uri'] + self.config['token'] + '/sendMessage'
self.chatid = self.config['chatid']
except AssertionError as error:
print("TELEGRAM_NOTIFIER: Configuration file not found. All methods will pass")
print(error)
self.config = None
pass
except yaml.YAMLError as error:
print("TELEGRAM_NOTIFIER: Cannot read configuration yaml. All methods will pass")
print(error)
self.config = None
pass
except KeyError as error:
print("TELEGRAM_NOTIFIER: Some key are missing. All methods will pass")
print(error)
self.config = None
pass
def post(self, m):
if self.config is None:
pass
assert type(m) is str, "Message must be a string"
payload = {'chat_id': self.chatid, 'text': ""}
for x in string(m).chunk(self.MAX_LENGTH):
payload['text'] = x
requests.post(self.uri_path, payload)
# How to use it:
t = TelegramNotifier()
t.post("This is a test message")
# /etc/telegram-notifier.conf
"""
---
uri: https://api.telegram.org/bot
token: YOUR_TOKEN
chatid: YOUR_CHET_ID
"""
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment