add telegram integraion

This commit is contained in:
2022-04-08 03:41:07 +03:00
parent 733c014229
commit 86af13f09e
12 changed files with 267 additions and 0 deletions

View File

@ -68,6 +68,9 @@ class Report:
if provider == ReportSettings.Console:
from ahriman.core.report.console import Console
return Console(architecture, configuration, section)
if provider == ReportSettings.Telegram:
from ahriman.core.report.telegram import Telegram
return Telegram(architecture, configuration, section)
return cls(architecture, configuration) # should never happen
def generate(self, packages: Iterable[Package], result: Result) -> None:

View File

@ -0,0 +1,93 @@
#
# Copyright (c) 2021-2022 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# technically we could use python-telegram-bot, but it is just a single request, cmon
import requests
from typing import Iterable
from ahriman.core.configuration import Configuration
from ahriman.core.report.jinja_template import JinjaTemplate
from ahriman.core.report.report import Report
from ahriman.core.util import exception_response_text
from ahriman.models.package import Package
from ahriman.models.result import Result
class Telegram(Report, JinjaTemplate):
"""
telegram report generator
:cvar TELEGRAM_API_URL: telegram api base url
:cvar TELEGRAM_MAX_CONTENT_LENGTH: max content length of the message
:ivar api_key: bot api key
:ivar chat_id: chat id to post message, either string with @ or integer
:ivar template_path: path to template for built packages
"""
TELEGRAM_API_URL = "https://api.telegram.org"
TELEGRAM_MAX_CONTENT_LENGTH = 4096
def __init__(self, architecture: str, configuration: Configuration, section: str) -> None:
"""
default constructor
:param architecture: repository architecture
:param configuration: configuration instance
:param section: settings section name
"""
Report.__init__(self, architecture, configuration)
JinjaTemplate.__init__(self, section, configuration)
self.api_key = configuration.get(section, "api_key")
self.chat_id = configuration.get(section, "chat_id")
self.template_path = configuration.getpath(section, "template_path")
def _send(self, text: str) -> None:
"""
send message to telegram channel
:param text: message body text
"""
try:
response = requests.post(
f"{self.TELEGRAM_API_URL}/bot{self.api_key}/sendMessage",
data={"chat_id": self.chat_id, "text": text})
response.raise_for_status()
except requests.HTTPError as e:
self.logger.exception("could not perform request: %s", exception_response_text(e))
raise
except Exception:
self.logger.exception("could not perform request")
raise
def generate(self, packages: Iterable[Package], result: Result) -> None:
"""
generate report for the specified packages
:param packages: list of packages to generate report
:param result: build result
"""
if not result.success:
return
text = self.make_html(result, self.template_path)
# telegram content is limited by 4096 symbols, so we are going to split the message by new lines
# to fit into this restriction
if len(text) > self.TELEGRAM_MAX_CONTENT_LENGTH:
position = text.rfind("\n", 0, self.TELEGRAM_MAX_CONTENT_LENGTH)
portion, text = text[:position], text[position + 1:] # +1 to exclude newline we split
self._send(portion)
# send remaining (or full in case if size is less than max length) text
self._send(text)

View File

@ -32,12 +32,14 @@ class ReportSettings(Enum):
:cvar HTML: html report generation
:cvar Email: email report generation
:cvar Console: print result to console
:cvar Telegram: markdown report to telegram channel
"""
Disabled = "disabled" # for testing purpose
HTML = "html"
Email = "email"
Console = "console"
Telegram = "telegram"
@classmethod
def from_option(cls: Type[ReportSettings], value: str) -> ReportSettings:
@ -52,4 +54,6 @@ class ReportSettings(Enum):
return cls.Email
if value.lower() in ("console",):
return cls.Console
if value.lower() in ("telegram",):
return cls.Telegram
raise InvalidOption(value)