mirror of
https://github.com/k2-fsa/icefall.git
synced 2025-08-08 09:32:20 +00:00
e63a8c27f8
* update manifest stats * update transcript configs * lang_char and compute_fbanks * save cuts in fbank_dir * add core codes * update decode.py * Create local/utils * tidy up * parse raw in prepare_lang_char.py * update manifest stats * update transcript configs * lang_char and compute_fbanks * save cuts in fbank_dir * add core codes * update decode.py * Create local/utils * tidy up * parse raw in prepare_lang_char.py * working train * Add compare_cer_transcript.py * fix tokenizer decode, allow d2f only * comment cleanup * add export files and READMEs * reword average column * fix comments * Update new results
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
import logging
|
|
from configparser import ConfigParser
|
|
|
|
import requests
|
|
|
|
|
|
def escape_html(text: str):
|
|
"""
|
|
Escapes all html characters in text
|
|
:param str text:
|
|
:rtype: str
|
|
"""
|
|
return text.replace("&", "&").replace("<", "<").replace(">", ">")
|
|
|
|
|
|
class TelegramStreamIO(logging.Handler):
|
|
|
|
API_ENDPOINT = "https://api.telegram.org"
|
|
MAX_MESSAGE_LEN = 4096
|
|
formatter = logging.Formatter(
|
|
"%(asctime)s - %(levelname)s at %(funcName)s "
|
|
"(line %(lineno)s):\n\n%(message)s"
|
|
)
|
|
|
|
def __init__(self, tg_configfile: str):
|
|
super(TelegramStreamIO, self).__init__()
|
|
config = ConfigParser()
|
|
if not config.read(tg_configfile):
|
|
raise FileNotFoundError(
|
|
f"{tg_configfile} not found. " "Retry without --telegram-cred flag."
|
|
)
|
|
config = config["TELEGRAM"]
|
|
token = config["token"]
|
|
self.chat_id = config["chat_id"]
|
|
self.url = f"{self.API_ENDPOINT}/bot{token}/sendMessage"
|
|
|
|
@staticmethod
|
|
def setup_logger(params):
|
|
if not params.telegram_cred:
|
|
return
|
|
formatter = logging.Formatter(
|
|
f"{params.exp_dir.name} %(asctime)s \n%(message)s"
|
|
)
|
|
tg = TelegramStreamIO(params.telegram_cred)
|
|
tg.setLevel(logging.WARN)
|
|
tg.setFormatter(formatter)
|
|
logging.getLogger("").addHandler(tg)
|
|
|
|
def emit(self, record: logging.LogRecord):
|
|
"""
|
|
Emit a record.
|
|
Send the record to the Web server as a percent-encoded dictionary
|
|
"""
|
|
data = {
|
|
"chat_id": self.chat_id,
|
|
"text": self.format(self.mapLogRecord(record)),
|
|
"parse_mode": "HTML",
|
|
}
|
|
try:
|
|
requests.get(self.url, json=data)
|
|
# return response.json()
|
|
except Exception as e:
|
|
logging.error(f"Failed to send telegram message: {repr(e)}")
|
|
pass
|
|
|
|
def mapLogRecord(self, record):
|
|
"""
|
|
Default implementation of mapping the log record into a dict
|
|
that is sent as the CGI data. Overwrite in your class.
|
|
Contributed by Franz Glasner.
|
|
"""
|
|
|
|
for k, v in record.__dict__.items():
|
|
if isinstance(v, str):
|
|
setattr(record, k, escape_html(v))
|
|
return record
|