# Copyright (c) 2016 The Pycroft Authors. See the AUTHORS file.
# This file is part of the Pycroft project and licensed under the terms of
# the Apache License, Version 2.0. See the LICENSE file for details.
"""
web
~~~~~~~~~~~~~~
This package contains the web interface based on flask
:copyright: (c) 2012 by AG DSN.
"""
import typing as t
import pathlib
from cmath import log
from datetime import datetime, date
from decimal import Decimal
from itertools import chain
from re import sub
import flask_babel
from flask import current_app, json, url_for, Flask
from jinja2 import pass_context
from jinja2.runtime import Context
from pycroft.helpers.i18n import localized, gettext
from pycroft.helpers.utc import ensure_tz
from pycroft.model import session
_filter_registry = {}
[docs]
def template_filter[_TF: t.Callable[..., t.Any]](name: str) -> t.Callable[[_TF], _TF]:
def decorator(fn: _TF) -> _TF:
_filter_registry[name] = fn
return fn
return decorator
_category_map = {"warning": "Warnung",
"error": "Fehler",
"info": "Hinweis",
"message": "Hinweis",
"success": "Erfolgreich"}
template_filter("localized")(localized)
[docs]
class AssetNotFound(Exception):
pass
# noinspection PyUnusedLocal
[docs]
@template_filter("require")
@pass_context
def require(ctx: Context, asset: str, **kwargs: t.Any) -> str:
"""
Build an URL for an asset generated by esbuild.
To prevent Jinja2 from inlining the calls of this filter with constant
parameters, this filter needs to be declared as a context filter. The
context is not actually used.
:param ctx: Template context
:param asset: Name of the
:param kwargs: Kwargs for :func:`url_for`
:return: URL
"""
asset_map, has_changed = current_app.extensions.get("esbuild", (None, None))
if asset_map is None or has_changed():
path = pathlib.Path(current_app.static_folder, 'manifest.json')
current_app.logger.info("Loading esbuild manifest from %s", path)
try:
mtime = path.stat().st_mtime
except FileNotFoundError as e:
raise RuntimeError(
"manifest.json not found. Did you forget"
" to execute esbuild? You might want to"
" take a look at the readme.md."
) from e
with path.open() as f:
asset_map = json.load(f)
def has_changed() -> bool:
try:
return path.stat().st_mtime != mtime
except OSError:
return False
current_app.extensions["esbuild"] = asset_map, has_changed
try:
filename = asset_map[asset]
except KeyError:
raise AssetNotFound(f"Asset {asset} not found") from None
kwargs["filename"] = filename
return url_for("static", **kwargs)
[docs]
@template_filter("pretty_category")
def pretty_category_filter(category: str) -> str:
"""Make pretty category names for flash messages, etc
"""
return _category_map.get(category, "Hinweis")
[docs]
@template_filter("date")
def date_filter(dt: datetime | date | None, format: str | None = None) -> str:
"""Format date or datetime objects using Flask-Babel
:param dt: a datetime object or None
:param format: format as understood by Flask-Babel's format_datetime
"""
if dt is None:
return "k/A"
return t.cast(str, flask_babel.format_date(dt, format))
[docs]
@template_filter("datetime")
def datetime_filter(dt: datetime | None, format: str | None = None) -> str:
"""Format datetime objects using Flask-Babel
:param dt: a datetime object or None
:param format: format as understood by Flask-Babel's format_datetime
"""
if dt is None:
return "k/A"
if isinstance(dt, str):
return dt
return t.cast(str, flask_babel.format_datetime(dt, format))
[docs]
@template_filter("timesince")
def timesince_filter(dt: datetime | None, default: str = "just now") -> str:
"""
Returns string representing "time since" e.g.
3 days ago, 5 hours ago etc.
adopted from
http://flask.pocoo.org/snippets/33/
"""
if dt is None:
return "k/A"
now = session.utcnow()
diff = now - ensure_tz(dt)
periods = (
(diff.days / 365, "Jahr", "Jahre"),
(diff.days / 30, "Monat", "Monate"),
(diff.days / 7, "Woche", "Wochen"),
(diff.days, "Tag", "Tage"),
(diff.seconds / 3600, "Stunde", "Stunden"),
(diff.seconds / 60, "Minute", "Minuten"),
(diff.seconds, "Sekunde", "Sekunden"),
)
return next(
(
f"vor {period:d} {singular if period == 1 else plural}"
for period, singular, plural in periods
),
default,
)
[docs]
def prefix_unit_filter(
value: float | Decimal, unit: str, factor: int, prefixes: t.Iterable[str]
) -> str:
units = list(chain(unit, (p + unit for p in prefixes)))
if value > 0:
n = min(int(log(value, factor).real), len(units)-1)
#todo change decimal formatting appropriately, previous {0:,4f} is wrong
return f"{float(value)/factor**n:,f} {units[n]}"
return f"0 {units[0]}"
[docs]
@template_filter("byte_size")
def byte_size_filter(value: float | Decimal) -> str:
return prefix_unit_filter(value, 'B', 1024, ['Ki', 'Mi', 'Gi', 'Ti'])
[docs]
@template_filter("money")
def money_filter(amount: float | Decimal) -> str:
return (f"{amount:.2f}\u202f€").replace('.', ',')
[docs]
@template_filter("icon")
def icon_filter(icon_class: str) -> str:
if len(tokens := icon_class.split(maxsplit=1)) == 2:
prefix, icon = tokens
else:
prefix = 'fas'
[icon] = tokens
return f"{prefix} {icon}"
[docs]
@template_filter("account_type")
def account_type_filter(account_type: str) -> str:
types = {
"USER_ASSET": gettext("User account (asset)"),
"BANK_ASSET": gettext("Bank account (asset)"),
"ASSET": gettext("Asset account"),
"LIABILITY": gettext("Liability account"),
"REVENUE": gettext("Revenue account"),
"EXPENSE": gettext("Expense account"),
"LEGACY": gettext("Legacy account"),
}
return types.get(account_type)
[docs]
@template_filter("transaction_type")
def transaction_type_filter(credit_debit_type: tuple[str, str]) -> str:
def remove_prefix(account_type_name: str) -> str:
return sub(r"[A-Z]+_(?=ASSET)", r"", account_type_name)
def replacer(types: tuple[str, str]) -> tuple[str, str] | None:
if not types:
return None
return (remove_prefix(types[0]), remove_prefix(types[1]))
types = {
("ASSET", "LIABILITY"): gettext("Balance sheet extension"),
("LIABILITY", "ASSET"): gettext("Balance sheet contraction"),
("ASSET", "REVENUE"): gettext("Revenue"),
("REVENUE", "ASSET"): gettext("Correcting entry (Revenue)"),
("EXPENSE", "ASSET"): gettext("Expense"),
("ASSET", "EXPENSE"): gettext("Correcting entry (Expense)"),
("ASSET", "ASSET"): gettext("Asset exchange"),
("LIABILITY", "LIABILITY"): gettext("Liability exchange")
}
return types.get(replacer(credit_debit_type), gettext("Unknown"))
"""
@template_filter("host_traffic")
def host_traffic_filter(host):
traffic_timespan = datetime.utcnow() - timedelta(days=7)
traffic_volumes = session.session.query(
TrafficVolume
).join(
TrafficVolume.ip
).join(
IP.host
).filter(
Host.id == host.id
).filter(
TrafficVolume.timestamp > traffic_timespan
).all()
traffic_sum = 0
for traffic in traffic_volumes:
traffic_sum += ( traffic.size / 1024 / 1024 )
return u"{} MB".format(traffic_sum)
"""
[docs]
def register_filters(app: Flask) -> None:
for name in _filter_registry:
app.jinja_env.filters[name] = _filter_registry[name]