import json import subprocess import urllib.request from collections import OrderedDict from typing import Optional, Tuple, List import paginate import sqlalchemy as sa from flask import request, abort, redirect, url_for, flash, render_template import wrpylib.mwmarkup import wrpylib.wrmwcache import wrpylib.wrmwmarkup from wradmin.app import db from wradmin.model import WrSledrunCache, WrReport from wradmin.template_helper import PylonsHelper from wrpylib.wrvalidators import rodelbahnbox_to_str, LonLat, difficulty_german_from_str, \ avalanches_german_from_str, public_transport_german_from_str, opt_tristate_german_comment_from_str, \ nightlightdays_from_str class RodelbahnController: def list(self): """Lists all sled runs""" q = db.session.query(WrSledrunCache) q = q.order_by(WrSledrunCache.page_title) paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)), url_maker=lambda page: url_for('rodelbahn_list', page=page), items_per_page=25) return render_template('rodelbahn_list.html', paginator=paginator) def view(self, id): """Displays a sled run""" q = db.session.query(WrSledrunCache) sledrun = q.get(id) if sledrun is None: abort(404) q = db.session.query(WrReport) q = q.filter_by(page_id=id).order_by(sa.sql.expression.desc(WrReport.id)) paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)), url_maker=lambda page: url_for('rodelbahn_view', id=id, page=page), items_per_page=25) return render_template('rodelbahn_view.html', sledding=sledrun, paginator=paginator) def view_wikitext(self, id): """Displays a sled run as MediaWiki wiki text""" q = db.session.query(WrSledrunCache) sledrun = q.get(id) if sledrun is None: abort(404) h = PylonsHelper() sledrun_json_url = h.sledrun_json_url(sledrun.page_title) sledrun_json_page = urllib.request.urlopen(sledrun_json_url) sledrun_json_str = sledrun_json_page.read() sledrun_json = json.loads(sledrun_json_str) def markdown_to_mediawiki(markdown: str) -> str: return subprocess.check_output(['pandoc', '--to', 'mediawiki'], input=markdown, encoding='utf-8') def position_to_lon_lat(value: Optional[dict]) -> Optional[LonLat]: if value is not None: lon = value.get('longitude') lat = value.get('latitude') if lon is not None and lat is not None: return LonLat(lon, lat) return None def position_ele_to_lon_lat(value: Optional[dict]) -> Optional[LonLat]: if value is not None: return position_to_lon_lat(value.get("position")) return None def position_ele_to_ele(value: Optional[dict]) -> Optional[int]: if value is not None: ele = value.get('elevation') if ele is not None: return int(ele) return None def aufstiegshilfe() -> Optional[List[Tuple[str, Optional[str]]]]: ws = sledrun_json.get('walkup_supports') if ws is None: return None return [(w['type'], w.get('comment')) for w in ws] def rodelverleih() -> Optional[List[Tuple[str, Optional[str]]]]: sr = sledrun_json.get('sled_rental_direct') if sr is None: return None return [('Ja', None)] if sr else [] def webauskunft() -> Tuple[Optional[bool], Optional[str]]: info_web = sledrun_json.get('info_web') if info_web is None: return None, None if len(info_web) == 0: return False, None return True, info_web[0]['url'] def telefonauskunft() -> Optional[List[Tuple[str, str]]]: info_phone = sledrun_json.get('info_phone') if info_phone is None: return None return [(pc['phone'], pc['name']) for pc in info_phone] def betreiber() -> str: has_operator = sledrun_json.get('has_operator') if has_operator is None: return sledrun_json.get('operator') if has_operator: return sledrun_json.get('operator') return 'Nein' sledrun_rbb_json = OrderedDict([ ('Position', position_to_lon_lat(sledrun_json.get('position'))), ('Position oben', position_ele_to_lon_lat(sledrun_json.get('top'))), ('Höhe oben', position_ele_to_ele(sledrun_json.get('top'))), ('Position unten', position_ele_to_lon_lat(sledrun_json.get('bottom'))), ('Höhe unten', position_ele_to_ele(sledrun_json.get('bottom'))), ('Länge', sledrun_json.get('length')), ('Schwierigkeit', difficulty_german_from_str(sledrun_json.get('difficulty', ''))), ('Lawinen', avalanches_german_from_str(sledrun_json.get('avalanches', ''))), ('Betreiber', betreiber()), ('Öffentliche Anreise', public_transport_german_from_str(sledrun_json.get('public_transport', ''))), ('Aufstieg möglich', sledrun_json.get('walkup_possible')), ('Aufstieg getrennt', opt_tristate_german_comment_from_str(sledrun_json.get('walkup_separate', ''))), ('Gehzeit', sledrun_json.get('walkup_time')), ('Aufstiegshilfe', aufstiegshilfe()), ('Beleuchtungsanlage', opt_tristate_german_comment_from_str(sledrun_json.get('nightlight_possible'))), ('Beleuchtungstage', nightlightdays_from_str(sledrun_json.get('nightlight_weekdays', ''))), ('Rodelverleih', rodelverleih()), ('Gütesiegel', None), ('Webauskunft', webauskunft()), ('Telefonauskunft', telefonauskunft()), ('Bild', sledrun_json.get('image')), ('In Übersichtskarte', sledrun_json.get('show_in_overview')), ('Forumid', sledrun_json.get('forum_id')) ]) def get_markdown_field(key: str) -> str: if key in sledrun_json: return markdown_to_mediawiki(sledrun_json[key]) return '' description = get_markdown_field('description').strip() night_light = get_markdown_field('night_light').strip() sled_rental_description = get_markdown_field('sled_rental_description').strip() rodelbahnbox = rodelbahnbox_to_str(sledrun_rbb_json) return render_template('rodelbahn_view_wikitext.html', sledrun=sledrun, sledrun_json=sledrun_json, rodelbahnbox=rodelbahnbox, description=description, night_light=night_light, sled_rental_description=sled_rental_description, operator=betreiber()) def json_edit(self, sledrun_id): q = db.session.query(WrSledrunCache) sledrun = q.get(sledrun_id) if sledrun is None: abort(404) h = PylonsHelper() schema_url = h.sledrun_json_schema_url() json_url = h.sledrun_json_url(sledrun.page_title) return render_template('json_editor.html', schema_url=schema_url, json_url=json_url) def update(self): """Updates the wrsledruncache table from the wiki""" c = db.session.connection() try: wrpylib.wrmwcache.update_wrsledruncache(c) db.session.commit() flash('Die Rodelbahnliste wurde erfolgreich aktualisiert.', 'info') except wrpylib.wrmwcache.UpdateCacheError as e: title = str(e.args[1]) title = wrpylib.mwmarkup.dbkey_to_title(title) msg = str(e.args[2]) msg = msg.replace('\n', '; ') if len(e.args) == 3: flash("Fehler bei Rodelbahn '{0}': {1}".format(title, msg), 'error') else: flash(str(e), 'error') # Redirect to result page return redirect(url_for('rodelbahn_list')) def update_regioncache(self): """Updates the wrregioncache table from the wiki""" c = db.session.connection() try: wrpylib.wrmwcache.update_wrregioncache(c) db.session.commit() flash('Die Rodelbahneinträge in den Regionslisten wurden erfolgreich aktualisiert.', 'info') except wrpylib.wrmwcache.UpdateCacheError as e: flash(str(e), 'error') # Redirect to result page return redirect(url_for('rodelbahn_list')) def update_mapcache(self): """Updates the wrmappointcache and wrmappathcache tables from the wiki.""" c = db.session.connection() try: wrpylib.wrmwcache.update_wrmapcache(c) db.session.commit() flash('Die Landkarteninformationen aus dem Wiki wurden erfolgreich aktualisiesrt.', 'info') except wrpylib.wrmwcache.UpdateCacheError as e: flash(str(e), 'error') # Redirect to result page return redirect(url_for('rodelbahn_list'))