-#!/usr/bin/python3.4
-from flask import request, abort, redirect, url_for, flash, render_template
+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.wrmwmarkup
-import wrpylib.wrmwcache
import wrpylib.mwmarkup
-
-import wradmin.model as model
+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 = model.meta.Session.query(model.WrSledrunCache)
- q = q.order_by(model.WrSledrunCache.page_title)
+ 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)
def view(self, id):
"""Displays a sled run"""
- q = model.meta.Session.query(model.WrSledrunCache)
- sledding = q.get(id)
- if sledding is None:
+ q = db.session.query(WrSledrunCache)
+ sledrun = q.get(id)
+ if sledrun is None:
abort(404)
- q = model.meta.Session.query(model.WrReport)
- q = q.filter_by(page_id=id).order_by(sa.sql.expression.desc(model.WrReport.id))
+ 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=sledding, paginator=paginator)
+ 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 = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrsledruncache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Rodelbahnliste wurde erfolgreich aktualisiert.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
title = str(e.args[1])
def update_regioncache(self):
"""Updates the wrregioncache table from the wiki"""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrregioncache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Rodelbahneinträge in den Regionslisten wurden erfolgreich aktualisiert.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
flash(str(e), 'error')
def update_mapcache(self):
"""Updates the wrmappointcache and wrmappathcache tables from the wiki."""
- c = model.meta.Session.connection()
+ c = db.session.connection()
try:
wrpylib.wrmwcache.update_wrmapcache(c)
- model.meta.Session.commit()
+ db.session.commit()
flash('Die Landkarteninformationen aus dem Wiki wurden erfolgreich aktualisiesrt.', 'info')
except wrpylib.wrmwcache.UpdateCacheError as e:
flash(str(e), 'error')
ToastFreeware / philipp / winterrodeln / wradmin.git / blobdiff