import re
import subprocess
from functools import cached_property
from io import BytesIO
from markupsafe import Markup
from onegov.core import utils
from onegov.core.orm.types import JSON
from onegov.core.static import StaticFile
from onegov.org import OrgApp
from onegov.org.app import get_common_asset as default_common_asset
from onegov.org.app import get_i18n_localedirs as get_org_i18n_localedirs
from onegov.org.models import GeneralFileCollection, GeneralFile
from onegov.winterthur.initial_content import create_new_organisation
from onegov.winterthur.roadwork import RoadworkClient
from onegov.winterthur.roadwork import RoadworkConfig
from onegov.winterthur.theme import WinterthurTheme
from pathlib import Path
from sqlalchemy import cast
from tempfile import TemporaryDirectory
from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable, Iterator
from onegov.core.cache import RedisCacheRegion
from onegov.org.models import Organisation
from onegov.winterthur.request import WinterthurRequest
from webob import Response
[docs]
class WinterthurApp(OrgApp):
[docs]
serve_static_files = True
[docs]
frame_ancestors = {
'https://winterthur.ch',
'https://*.winterthur.ch',
'http://localhost:8000',
}
# disable same site cookie protection as we need to run inside iframes
# with cookies enabled
[docs]
same_site_cookie_policy = None
[docs]
def enable_iframes(self, request: 'WinterthurRequest') -> None:
request.content_security_policy.frame_ancestors |= self.frame_ancestors
request.include('iframe-resizer')
@property
[docs]
def roadwork_cache(self) -> 'RedisCacheRegion':
# the expiration time is high here, as the expiration is more closely
# managed by the roadwork client
return self.get_cache('roadwork', expiration_time=60 * 60 * 24)
@cached_property
[docs]
def roadwork_client(self) -> RoadworkClient:
config = RoadworkConfig.lookup()
# FIXME: We should probably raise an error if the configuration
# is incomplete...
return RoadworkClient(
cache=self.roadwork_cache,
hostname=config.hostname, # type:ignore[arg-type]
endpoint=config.endpoint,
username=config.username, # type:ignore[arg-type]
password=config.password # type:ignore[arg-type]
)
@property
[docs]
def mission_report_legend(self) -> Markup:
from onegov.winterthur.views.settings import DEFAULT_LEGEND
settings = self.org.meta.get('mission_report_settings') or {}
if 'legend' in settings:
# NOTE: We need to wrap this in Markup. It would be cleaner
# if we had a proxy settings object with dict_property
return Markup(settings['legend']) # noqa: MS001
return DEFAULT_LEGEND
@property
[docs]
def hide_civil_defence_field(self) -> bool:
settings = self.org.meta.get('mission_report_settings') or {}
hide = settings.get('hide_civil_defence_field', False)
return hide
[docs]
def static_file(self, path: str) -> StaticFile:
return StaticFile(path, version=self.version)
[docs]
def get_shift_schedule_image(self) -> BytesIO | None:
""" Gets or creates an image of the latest public pdf.
We store the image using the last modified timestamp - this way, we
have a version of past images. Note that we don't delete any old
images of shift schedules.
"""
query = GeneralFileCollection(self.session()).query().filter(
GeneralFile.published.is_(True),
cast(GeneralFile.reference, JSON)['content_type']
== 'application/pdf'
)
query = query.order_by(GeneralFile.created.desc())
file = query.first()
if not file:
return None
upload_time = file.created.timestamp()
filename = f'shift-schedule-{upload_time}.png'
fs = self.filestorage
assert fs is not None
if not fs.exists(filename):
with TemporaryDirectory() as directory:
path = Path(directory)
with (path / 'input.pdf').open('wb') as pdf:
pdf.write(file.reference.file.read())
process = subprocess.run((
'gs',
# disable read/writes outside of the given files
'-dSAFER',
'-dPARANOIDSAFER',
# do not block for any reason
'-dBATCH',
'-dNOPAUSE',
'-dNOPROMPT',
# limit output messages
'-dQUIET',
'-sstdout=/dev/null',
# format the page for thumbnails
'-dPDFFitPage',
# render in high resolution before downscaling to 300 dpi
f'-r{300}',
f'-dDownScaleFactor={1}',
# only use the first page
'-dLastPage=1',
# output to png
'-sDEVICE=png16m',
f'-sOutputFile={path / "preview.png"}',
# force landscape orientation in postscript
'-c',
'<</Orientation 3>> setpagedevice',
'-f',
# from pdf
str(path / 'input.pdf')
))
process.check_returncode()
with (
(path / 'preview.png').open('rb') as input,
fs.open(filename, 'wb') as output
):
# NOTE: Bug in type hints of FS
output.write(input.read()) # type:ignore
with fs.open(filename, 'rb') as input:
# NOTE: Bug in type hints of FS
return BytesIO(input.read()) # type:ignore
@WinterthurApp.tween_factory()
[docs]
def enable_iframes_tween_factory(
app: WinterthurApp,
handler: 'Callable[[WinterthurRequest], Response]'
) -> 'Callable[[WinterthurRequest], Response]':
iframe_paths = (
r'/streets.*',
r'/director(y|ies|y-submission/.*)',
r'/ticket/.*',
r'/mission-report.*',
r'/roadwork.*',
r'/daycare-subsidy-calculator',
r'/events.*',
r'/event.*',
)
iframe_path_re = re.compile(rf"({'|'.join(iframe_paths)})")
def enable_iframes_tween(request: 'WinterthurRequest') -> 'Response':
""" Enables iframes on matching paths. """
result = handler(request)
if iframe_path_re.match(request.path_info or ''):
request.app.enable_iframes(request)
return result
return enable_iframes_tween
@WinterthurApp.template_directory()
[docs]
def get_template_directory() -> str:
return 'templates'
@OrgApp.static_directory()
[docs]
def get_static_directory() -> str:
return 'static'
@WinterthurApp.setting(section='core', name='theme')
[docs]
def get_theme() -> WinterthurTheme:
return WinterthurTheme()
@WinterthurApp.setting(section='org', name='create_new_organisation')
[docs]
def get_create_new_organisation_factory(
) -> 'Callable[[WinterthurApp, str], Organisation]':
return create_new_organisation
@WinterthurApp.setting(section='org', name='default_directory_search_widget')
@WinterthurApp.setting(section='org', name='default_event_search_widget')
@WinterthurApp.setting(section='i18n', name='localedirs')
[docs]
def get_i18n_localedirs() -> list[str]:
mine = utils.module_path('onegov.winterthur', 'locale')
return [mine, *get_org_i18n_localedirs()]
@WinterthurApp.webasset_path()
[docs]
def get_js_path() -> str:
return 'assets/js'
@WinterthurApp.webasset_output()
[docs]
def get_webasset_output() -> str:
return 'assets/bundles'
@WinterthurApp.webasset('street-search')
[docs]
def get_search_asset() -> 'Iterator[str]':
yield 'wade.js'
yield 'string-score.js'
yield 'street-search.js'
@WinterthurApp.webasset('iframe-resizer')
[docs]
def get_iframe_resizer() -> 'Iterator[str]':
yield 'iframe-resizer-options.js'
yield 'iframe-resizer-contentwindow.js'
@WinterthurApp.webasset('iframe-enhancements')
[docs]
def get_iframe_enhancements() -> 'Iterator[str]':
yield 'iframe-enhancements.js'
@WinterthurApp.webasset('common')
[docs]
def get_common_asset() -> 'Iterator[str]':
yield from default_common_asset()
yield 'winterthur.js'