Last active
June 9, 2016 10:08
-
-
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.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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 |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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'); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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