arcanis/ahriman
1
0
Fork 0
mirror of https://github.com/arcan1s/ahriman.git synced 2025-07-22 02:09:56 +00:00
Code Issues Projects Releases Wiki Activity

feat: implement rss generation (#130)

Browse Source
This commit is contained in:
Evgenii Alekseev
2024-08-29 16:53:40 +03:00
parent d7c4fccf98
commit 529d4caa0e
18 changed files with 481 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/ahriman/core/build_tools/package_archive.py
View File

@ -171,7 +171,7 @@ class PackageArchive:
result: dict[Path, list[FilesystemPackage]] = {}
# sort items from children directories to root
for path, packages in reversed(sorted(source.items())):
for path, packages in sorted(source.items(), reverse=True):
# skip if this path belongs to the one of the base packages
if any(package.package_name in base_packages for package in packages):
continue

53
src/ahriman/core/report/jinja_template.py
View File

@ -17,14 +17,16 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import datetime
import jinja2
from collections.abc import Callable
from pathlib import Path
from typing import Any
from ahriman.core.configuration import Configuration
from ahriman.core.sign.gpg import GPG
from ahriman.core.utils import pretty_datetime, pretty_size
from ahriman.core.utils import pretty_datetime, pretty_size, utcnow
from ahriman.models.repository_id import RepositoryId
from ahriman.models.result import Result
from ahriman.models.sign_settings import SignSettings
@ -37,6 +39,7 @@ class JinjaTemplate:
It uses jinja2 templates for report generation, the following variables are allowed:
* homepage - link to homepage, string, optional
* last_update - report generation time, pretty printed datetime, required
* link_path - prefix fo packages to download, string, required
* has_package_signed - ``True`` in case if package sign enabled, ``False`` otherwise, required
* has_repo_signed - ``True`` in case if repository database sign enabled, ``False`` otherwise, required
@ -46,21 +49,24 @@ class JinjaTemplate:
* build_date, pretty printed datetime, string
* depends, sorted list of strings
* description, string
* filename, string,
* filename, string
* groups, sorted list of strings
* installed_size, pretty printed datetime, string
* installed_size, pretty printed size, string
* licenses, sorted list of strings
* name, string
* tag, string
* url, string
* version, string
* pgp_key - default PGP key ID, string, optional
* repository - repository name, string, required
* rss_url - optional link to the RSS feed, string, optional
Attributes:
default_pgp_key(str | None): default PGP key
homepage(str | None): homepage link if any (for footer)
link_path(str): prefix fo packages to download
name(str): repository name
rss_url(str | None): link to the RSS feed
sign_targets(set[SignSettings]): targets to sign enabled in configuration
templates(list[Path]): list of directories with templates
"""
@ -80,8 +86,36 @@ class JinjaTemplate:
self.homepage = configuration.get(section, "homepage", fallback=None)
self.link_path = configuration.get(section, "link_path")
self.name = repository_id.name
self.rss_url = configuration.get(section, "rss_url", fallback=None)
self.sign_targets, self.default_pgp_key = GPG.sign_options(configuration)
@staticmethod
def format_datetime(timestamp: datetime.datetime | float | int | None) -> str:
"""
convert datetime object to string
Args:
timestamp(datetime.datetime | float | int | None): datetime to convert
Returns:
str: datetime as string representation
"""
return pretty_datetime(timestamp)
@staticmethod
def sort_content(content: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""
sort content before rendering
Args:
content(list[dict[str, str]]): content of the template
Returns:
list[dict[str, str]]: sorted content according to comparator defined
"""
comparator: Callable[[dict[str, str]], str] = lambda item: item["filename"]
return sorted(content, key=comparator)
def make_html(self, result: Result, template_name: Path | str) -> str:
"""
generate report for the specified packages
@ -104,7 +138,7 @@ class JinjaTemplate:
{
"architecture": properties.architecture or "",
"archive_size": pretty_size(properties.archive_size),
"build_date": pretty_datetime(properties.build_date),
"build_date": self.format_datetime(properties.build_date),
"depends": properties.depends,
"description": properties.description or "",
"filename": properties.filename,
@ -112,17 +146,20 @@ class JinjaTemplate:
"installed_size": pretty_size(properties.installed_size),
"licenses": properties.licenses,
"name": package,
"tag": f"tag:{self.name}:{properties.architecture}:{package}:{base.version}:{properties.build_date}",
"url": properties.url or "",
"version": base.version
"version": base.version,
} for base in result.success for package, properties in base.packages.items()
]
comparator: Callable[[dict[str, str]], str] = lambda item: item["filename"]
return template.render(
homepage=self.homepage,
last_update=self.format_datetime(utcnow()),
link_path=self.link_path,
has_package_signed=SignSettings.Packages in self.sign_targets,
has_repo_signed=SignSettings.Repository in self.sign_targets,
packages=sorted(content, key=comparator),
packages=self.sort_content(content),
pgp_key=self.default_pgp_key,
repository=self.name)
repository=self.name,
rss_url=self.rss_url,
)

5
src/ahriman/core/report/report.py
View File

@ -66,7 +66,7 @@ class Report(LazyLogging):
self.configuration = configuration
@staticmethod
def load(repository_id: RepositoryId, configuration: Configuration, target: str) -> Report:
def load(repository_id: RepositoryId, configuration: Configuration, target: str) -> Report: # pylint: disable=too-many-return-statements
"""
load client from settings
@ -92,6 +92,9 @@ class Report(LazyLogging):
case ReportSettings.Telegram:
from ahriman.core.report.telegram import Telegram
return Telegram(repository_id, configuration, section)
case ReportSettings.RSS:
from ahriman.core.report.rss import RSS
return RSS(repository_id, configuration, section)
case ReportSettings.RemoteCall:
from ahriman.core.report.remote_call import RemoteCall
return RemoteCall(repository_id, configuration, section)

77
src/ahriman/core/report/report_trigger.py
View File

@ -116,6 +116,11 @@ class ReportTrigger(Trigger):
"required": True,
"empty": False,
},
"rss_url": {
"type": "string",
"empty": False,
"is_url": ["http", "https"],
},
"sender": {
"type": "string",
"required": True,
@ -187,6 +192,11 @@ class ReportTrigger(Trigger):
"coerce": "absolute_path",
"required": True,
},
"rss_url": {
"type": "string",
"empty": False,
"is_url": ["http", "https"],
},
"template": {
"type": "string",
"excludes": ["template_path"],
@ -243,6 +253,11 @@ class ReportTrigger(Trigger):
"empty": False,
"is_url": [],
},
"rss_url": {
"type": "string",
"empty": False,
"is_url": ["http", "https"],
},
"template": {
"type": "string",
"excludes": ["template_path"],
@ -304,7 +319,67 @@ class ReportTrigger(Trigger):
"coerce": "integer",
},
},
}
},
"rss": {
"type": "dict",
"schema": {
"type": {
"type": "string",
"allowed": ["rss"],
},
"homepage": {
"type": "string",
"empty": False,
"is_url": ["http", "https"],
},
"link_path": {
"type": "string",
"required": True,
"empty": False,
"is_url": [],
},
"max_entries": {
"type": "integer",
"coerce": "integer",
},
"path": {
"type": "path",
"coerce": "absolute_path",
"required": True,
},
"rss_url": {
"type": "string",
"empty": False,
"is_url": ["http", "https"],
},
"template": {
"type": "string",
"excludes": ["template_path"],
"dependencies": ["templates"],
"required": True,
"empty": False,
},
"template_path": {
"type": "path",
"coerce": "absolute_path",
"excludes": ["template"],
"required": True,
"path_exists": True,
"path_type": "file",
},
"templates": {
"type": "list",
"coerce": "list",
"schema": {
"type": "path",
"coerce": "absolute_path",
"path_exists": True,
"path_type": "dir",
},
"empty": False,
},
},
},
}
def __init__(self, repository_id: RepositoryId, configuration: Configuration) -> None:

130
src/ahriman/core/report/rss.py Normal file
View File

@ -0,0 +1,130 @@
#
# Copyright (c) 2021-2024 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/>.
#
import datetime
from collections.abc import Callable
from email.utils import format_datetime, parsedate_to_datetime
from typing import Any
from ahriman.core import context
from ahriman.core.configuration import Configuration
from ahriman.core.report.jinja_template import JinjaTemplate
from ahriman.core.report.report import Report
from ahriman.core.status import Client
from ahriman.models.event import EventType
from ahriman.models.package import Package
from ahriman.models.repository_id import RepositoryId
from ahriman.models.result import Result
class RSS(Report, JinjaTemplate):
"""
RSS report generator
Attributes:
max_entries(int): the maximal amount of entries in RSS
report_path(Path): output path to RSS report
template(str): name of the template
"""
def __init__(self, repository_id: RepositoryId, configuration: Configuration, section: str) -> None:
"""
default constructor
Args:
repository_id(RepositoryId): repository unique identifier
configuration(Configuration): configuration instance
section(str): settings section name
"""
Report.__init__(self, repository_id, configuration)
JinjaTemplate.__init__(self, repository_id, configuration, section)
self.max_entries = configuration.getint(section, "max_entries", fallback=-1)
self.report_path = configuration.getpath(section, "path")
self.template = configuration.get(section, "template")
@staticmethod
def format_datetime(timestamp: datetime.datetime | float | int | None) -> str:
"""
convert datetime object to string
Args:
timestamp(datetime.datetime | float | int | None): datetime to convert
Returns:
str: datetime as string representation
"""
if timestamp is None:
return ""
if isinstance(timestamp, (int, float)):
timestamp = datetime.datetime.fromtimestamp(timestamp, datetime.UTC)
return format_datetime(timestamp)
@staticmethod
def sort_content(content: list[dict[str, Any]]) -> list[dict[str, Any]]:
"""
sort content before rendering
Args:
content(list[dict[str, str]]): content of the template
Returns:
list[dict[str, str]]: sorted content according to comparator defined
"""
comparator: Callable[[dict[str, str]], datetime.datetime] = \
lambda item: parsedate_to_datetime(item["build_date"])
return sorted(content, key=comparator, reverse=True)
def content(self, packages: list[Package]) -> Result:
"""
extract result to be written to template
Args:
packages(list[Package]): list of packages to generate report
Returns:
Result: result descriptor
"""
ctx = context.get()
reporter = ctx.get(Client)
events = reporter.event_get(EventType.PackageUpdated, None, limit=self.max_entries)
known_packages = {package.base: package for package in packages}
result = Result()
for event in events:
package = known_packages.get(event.object_id)
if package is None:
continue # package not found
result.add_updated(package)
return result
def generate(self, packages: list[Package], result: Result) -> None:
"""
generate report for the specified packages
Args:
packages(list[Package]): list of packages to generate report
result(Result): build result
"""
result = self.content(packages)
rss = self.make_html(result, self.template)
self.report_path.write_text(rss, encoding="utf8")

6
src/ahriman/models/report_settings.py
View File

@ -32,6 +32,7 @@ class ReportSettings(StrEnum):
Email(ReportSettings): (class attribute) email report generation
Console(ReportSettings): (class attribute) print result to console
Telegram(ReportSettings): (class attribute) markdown report to telegram channel
RSS(ReportSettings): (class attribute) RSS report generation
RemoteCall(ReportSettings): (class attribute) remote ahriman server call
"""
@ -40,10 +41,11 @@ class ReportSettings(StrEnum):
Email = "email"
Console = "console"
Telegram = "telegram"
RSS = "rss"
RemoteCall = "remote-call"
@staticmethod
def from_option(value: str) -> ReportSettings:
def from_option(value: str) -> ReportSettings: # pylint: disable=too-many-return-statements
"""
construct value from configuration
@ -62,6 +64,8 @@ class ReportSettings(StrEnum):
return ReportSettings.Console
case "telegram":
return ReportSettings.Telegram
case "rss":
return ReportSettings.RSS
case "ahriman" | "remote-call":
return ReportSettings.RemoteCall
case _: