wradmin/controllers/rodelbahn.py

   1 import json
   2 import subprocess
   3 import urllib.request
   4 from collections import OrderedDict
   5 from typing import Optional, Tuple, List
   6
   7 import paginate
   8 import sqlalchemy as sa
   9 from flask import request, abort, redirect, url_for, flash, render_template
  10
  11 import wrpylib.mwmarkup
  12 import wrpylib.wrmwcache
  13 import wrpylib.wrmwmarkup
  14 from wradmin.app import db
  15 from wradmin.model import WrSledrunCache, WrReport
  16 from wradmin.template_helper import PylonsHelper
  17 from wrpylib.wrvalidators import rodelbahnbox_to_str, LonLat, difficulty_german_from_str, \
  18     avalanches_german_from_str, public_transport_german_from_str, opt_tristate_german_comment_from_str, \
  19     nightlightdays_from_str
  20
  21
  22 class RodelbahnController:
  23
  24     def list(self):
  25         """Lists all sled runs"""
  26         q = db.session.query(WrSledrunCache)
  27         q = q.order_by(WrSledrunCache.page_title)
  28         paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
  29                                   url_maker=lambda page: url_for('rodelbahn_list', page=page),
  30                                   items_per_page=25)
  31         return render_template('rodelbahn_list.html', paginator=paginator)
  32
  33     def view(self, id):
  34         """Displays a sled run"""
  35         q = db.session.query(WrSledrunCache)
  36         sledrun = q.get(id)
  37         if sledrun is None:
  38             abort(404)
  39         q = db.session.query(WrReport)
  40         q = q.filter_by(page_id=id).order_by(sa.sql.expression.desc(WrReport.id))
  41         paginator = paginate.Page(q.all(), page=int(request.args.get('page', 1)),
  42                                   url_maker=lambda page: url_for('rodelbahn_view', id=id, page=page),
  43                                   items_per_page=25)
  44         return render_template('rodelbahn_view.html', sledding=sledrun, paginator=paginator)
  45
  46     def view_wikitext(self, id):
  47         """Displays a sled run as MediaWiki wiki text"""
  48         q = db.session.query(WrSledrunCache)
  49         sledrun = q.get(id)
  50         if sledrun is None:
  51             abort(404)
  52         h = PylonsHelper()
  53         sledrun_json_url = h.sledrun_json_url(sledrun.page_title)
  54         sledrun_json_page = urllib.request.urlopen(sledrun_json_url)
  55         sledrun_json_str = sledrun_json_page.read()
  56         sledrun_json = json.loads(sledrun_json_str)
  57
  58         def markdown_to_mediawiki(markdown: str) -> str:
  59             return subprocess.check_output(['pandoc', '--to', 'mediawiki'], input=markdown, encoding='utf-8')
  60
  61         def position_to_lon_lat(value: Optional[dict]) -> Optional[LonLat]:
  62             if value is not None:
  63                 lon = value.get('longitude')
  64                 lat = value.get('latitude')
  65                 if lon is not None and lat is not None:
  66                     return LonLat(lon, lat)
  67             return None
  68
  69         def position_ele_to_lon_lat(value: Optional[dict]) -> Optional[LonLat]:
  70             if value is not None:
  71                 return position_to_lon_lat(value.get("position"))
  72             return None
  73
  74         def position_ele_to_ele(value: Optional[dict]) -> Optional[int]:
  75             if value is not None:
  76                 ele = value.get('elevation')
  77                 if ele is not None:
  78                     return int(ele)
  79             return None
  80
  81         def aufstiegshilfe() -> Optional[List[Tuple[str, Optional[str]]]]:
  82             ws = sledrun_json.get('walkup_supports')
  83             if ws is None:
  84                 return None
  85             return [(w['type'], w.get('comment')) for w in ws]
  86
  87         def rodelverleih() -> Optional[List[Tuple[str, Optional[str]]]]:
  88             sr = sledrun_json.get('sled_rental_direct')
  89             if sr is None:
  90                 return None
  91             return [('Ja', None)] if sr else []
  92
  93         def webauskunft() -> Tuple[Optional[bool], Optional[str]]:
  94             info_web = sledrun_json.get('info_web')
  95             if info_web is None:
  96                 return None, None
  97             if len(info_web) == 0:
  98                 return False, None
  99             return True, info_web[0]['url']
 100
 101         def telefonauskunft() -> Optional[List[Tuple[str, str]]]:
 102             info_phone = sledrun_json.get('info_phone')
 103             if info_phone is None:
 104                 return None
 105             return [(pc['phone'], pc['name']) for pc in info_phone]
 106
 107         def betreiber() -> str:
 108             has_operator = sledrun_json.get('has_operator')
 109             if has_operator is None:
 110                 return sledrun_json.get('operator')
 111             if has_operator:
 112                 return sledrun_json.get('operator')
 113             return 'Nein'
 114
 115         sledrun_rbb_json = OrderedDict([
 116             ('Position', position_to_lon_lat(sledrun_json.get('position'))),
 117             ('Position oben', position_ele_to_lon_lat(sledrun_json.get('top'))),
 118             ('Höhe oben', position_ele_to_ele(sledrun_json.get('top'))),
 119             ('Position unten', position_ele_to_lon_lat(sledrun_json.get('bottom'))),
 120             ('Höhe unten', position_ele_to_ele(sledrun_json.get('bottom'))),
 121             ('Länge', sledrun_json.get('length')),
 122             ('Schwierigkeit', difficulty_german_from_str(sledrun_json.get('difficulty', ''))),
 123             ('Lawinen', avalanches_german_from_str(sledrun_json.get('avalanches', ''))),
 124             ('Betreiber', betreiber()),
 125             ('Öffentliche Anreise', public_transport_german_from_str(sledrun_json.get('public_transport', ''))),
 126             ('Aufstieg möglich', sledrun_json.get('walkup_possible')),
 127             ('Aufstieg getrennt', opt_tristate_german_comment_from_str(sledrun_json.get('walkup_separate', ''))),
 128             ('Gehzeit', sledrun_json.get('walkup_time')),
 129             ('Aufstiegshilfe', aufstiegshilfe()),
 130             ('Beleuchtungsanlage', opt_tristate_german_comment_from_str(sledrun_json.get('nightlight_possible'))),
 131             ('Beleuchtungstage', nightlightdays_from_str(sledrun_json.get('nightlight_weekdays', ''))),
 132             ('Rodelverleih', rodelverleih()),
 133             ('Gütesiegel', None),
 134             ('Webauskunft', webauskunft()),
 135             ('Telefonauskunft', telefonauskunft()),
 136             ('Bild', sledrun_json.get('image')),
 137             ('In Übersichtskarte', sledrun_json.get('show_in_overview')),
 138             ('Forumid', sledrun_json.get('forum_id'))
 139         ])
 140
 141         def get_markdown_field(key: str) -> str:
 142             if key in sledrun_json:
 143                 return markdown_to_mediawiki(sledrun_json[key])
 144             return ''
 145
 146         description = get_markdown_field('description').strip()
 147         night_light = get_markdown_field('night_light').strip()
 148         sled_rental_description = get_markdown_field('sled_rental_description').strip()
 149
 150         rodelbahnbox = rodelbahnbox_to_str(sledrun_rbb_json)
 151         return render_template('rodelbahn_view_wikitext.html', sledrun=sledrun, sledrun_json=sledrun_json,
 152                                rodelbahnbox=rodelbahnbox, description=description, night_light=night_light,
 153                                sled_rental_description=sled_rental_description, operator=betreiber())
 154
 155     def json_edit(self, sledrun_id):
 156         q = db.session.query(WrSledrunCache)
 157         sledrun = q.get(sledrun_id)
 158         if sledrun is None:
 159             abort(404)
 160         h = PylonsHelper()
 161         schema_url = h.sledrun_json_schema_url()
 162         json_url = h.sledrun_json_url(sledrun.page_title)
 163         return render_template('json_editor.html', schema_url=schema_url, json_url=json_url)
 164
 165     def update(self):
 166         """Updates the wrsledruncache table from the wiki"""
 167         c = db.session.connection()
 168         try:
 169             wrpylib.wrmwcache.update_wrsledruncache(c)
 170             db.session.commit()
 171             flash('Die Rodelbahnliste wurde erfolgreich aktualisiert.', 'info')
 172         except wrpylib.wrmwcache.UpdateCacheError as e:
 173             title = str(e.args[1])
 174             title = wrpylib.mwmarkup.dbkey_to_title(title)
 175             msg = str(e.args[2])
 176             msg = msg.replace('\n', '; ')
 177             if len(e.args) == 3:
 178                 flash("Fehler bei Rodelbahn '{0}': {1}".format(title, msg), 'error')
 179             else:
 180                 flash(str(e), 'error')
 181         # Redirect to result page
 182         return redirect(url_for('rodelbahn_list'))
 183
 184     def update_regioncache(self):
 185         """Updates the wrregioncache table from the wiki"""
 186         c = db.session.connection()
 187         try:
 188             wrpylib.wrmwcache.update_wrregioncache(c)
 189             db.session.commit()
 190             flash('Die Rodelbahneinträge in den Regionslisten wurden erfolgreich aktualisiert.', 'info')
 191         except wrpylib.wrmwcache.UpdateCacheError as e:
 192             flash(str(e), 'error')
 193         # Redirect to result page
 194         return redirect(url_for('rodelbahn_list'))
 195
 196     def update_mapcache(self):
 197         """Updates the wrmappointcache and wrmappathcache tables from the wiki."""
 198         c = db.session.connection()
 199         try:
 200             wrpylib.wrmwcache.update_wrmapcache(c)
 201             db.session.commit()
 202             flash('Die Landkarteninformationen aus dem Wiki wurden erfolgreich aktualisiesrt.', 'info')
 203         except wrpylib.wrmwcache.UpdateCacheError as e:
 204             flash(str(e), 'error')
 205         # Redirect to result page
 206         return redirect(url_for('rodelbahn_list'))