#!/usr/bin/python """ Update sledrun JSON files based on OpenData published at https://www.data.gv.at/katalog/dataset/land-tirol_rodelbahnenintirol At the time of writing, the OpenData were be outdated - there is another official list: https://www.tirol.gv.at/sport/richtlinien-und-initiativen/sport-guetesiegel/naturrodelbahnguetesiegel/rodelbahnen/ Get list of existing Gütesiegel entries: SELECT page_title, json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.name') AS name, json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.stätte_id') AS staette_id, aswkt( point(json_extract(sledrun_json, '$.position.longitude'), json_extract(sledrun_json, '$.position.latitude'))) AS position, json_extract(sledrun_json, '$.entry_under_construction') AS in_arbeit FROM wrsledrunjsoncache JOIN page ON sledrun_page_id=page_id WHERE json_exists(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel') ORDER BY page_title; Get existing Gütesiegel entry with matching stätte_id: SELECT COUNT(*) FROM wrsledrunjsoncache WHERE json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.stätte_id') = 3344; Get list of sledruns closer to a given point (longitude, latitude) - distance is in meter: SELECT page_title, name, aswkt(position) as position, in_arbeit, distance FROM ( SELECT *, st_distance_sphere(point(11.4, 47.3), position) as distance FROM ( SELECT page_title, json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.name') AS name, point(json_extract(sledrun_json, '$.position.longitude'), json_extract(sledrun_json, '$.position.latitude')) AS position, json_extract(sledrun_json, '$.entry_under_construction') AS in_arbeit FROM wrsledrunjsoncache JOIN page ON sledrun_page_id=page_id WHERE json_exists(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel') ) AS g ) AS d WHERE distance < 20000 ORDER BY distance; """ import argparse import configparser import json from typing import List, Union import fiona import requests import sqlalchemy from fiona.transform import transform_geom from requests import Session from shapely.geometry import shape from sqlalchemy import create_engine, text from wrpylib.json_tools import format_json def update_sledrun(connection: sqlalchemy.engine.Connection, api_url: str, session: Session, csrf_token: str, shp_crs: Union[dict, str], lon_lat_crs: Union[dict, str], feature: dict): # Is a sledrun already existing with the given stätte_id? properties = feature['properties'] sledrun_type = properties['ATTR_TYP'] assert sledrun_type in ['Naturrodelbahn', 'Sommerrodelbahn'] if sledrun_type != 'Naturrodelbahn': return count, = connection.execute(text(""" SELECT COUNT(*) FROM wrsledrunjsoncache WHERE json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.stätte_id') = :staette_id;"""), [{'staette_id': properties['STAETTE_ID']}]).fetchone() if count == 0: # Is a sledrun already existing with the given name? rows = list(connection.execute(text(""" SELECT sledrun_page_id, sledrun_json_page_id, sledrun_json_rev_id, sledrun_json FROM wrsledrunjsoncache WHERE json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.name') = :name;"""), [{'name': properties['STAETTE_NA']}])) if len(rows) == 0: print(f"{properties['STAETTE_NA']} ({properties['STAETTE_ID']}) not found") # Is there a sledrun that's close? line_string_lon_lat = transform_geom(shp_crs, lon_lat_crs, feature['geometry']) point = shape(line_string_lon_lat).representative_point() lon, lat = point.x, point.y candidates = connection.execute(text(""" SELECT page_title, name, aswkt(position) as position, in_arbeit, distance FROM ( SELECT *, st_distance_sphere(point(:lon, :lat), position) as distance FROM ( SELECT page_title, json_extract(sledrun_json, '$.tiroler_naturrodelbahn_gütesiegel.name') AS name, point(json_extract(sledrun_json, '$.position.longitude'), json_extract(sledrun_json, '$.position.latitude')) AS position, json_extract(sledrun_json, '$.entry_under_construction') AS in_arbeit FROM wrsledrunjsoncache JOIN page ON sledrun_page_id=page_id ) AS g ) AS d WHERE distance < 5000 ORDER BY distance; """), [{'lon': lon, 'lat': lat}]) candidates = list(candidates) if len(candidates) == 0: print(' No candidates.') print(' Candidates:') for candidate in candidates: print(f' {candidate}') elif len(rows) == 1: print(f"{properties['STAETTE_NA']} -> {properties['STAETTE_ID']} (match by name)") # add 'stätte_id' to JSON row = rows[0] sledrun_json = json.loads(row.sledrun_json) gs = sledrun_json['tiroler_naturrodelbahn_gütesiegel'] if gs.get('stätte_id') is not None: print(f"{properties['STAETTE_NA']} has a different ID") return gs['stätte_id'] = properties['STAETTE_ID'] new_sledrun_json = format_json(sledrun_json) post_data = { 'action': 'edit', 'pageid': row.sledrun_json_page_id, 'text': new_sledrun_json, 'summary': '"Stätte-ID" von Tiroler Naturrodelbahn-Gütesiegel eingefügt.', 'minor': 1, 'bot': 1, 'baserevid': row.sledrun_json_rev_id, 'nocreate': 1, 'contentformat': 'application/json', 'contentmodel': 'json', 'token': csrf_token, 'format': 'json', } response = session.post(url=api_url, data=post_data) response = response.json()['edit'] if response['result'] != 'Success': raise RuntimeError(response['reason']) assert row.sledrun_json_page_id == response['pageid'] if 'nochange' in response: print("Change already done previously - table wrsledrunjsoncache doesn't seem to be up-to-date.") return assert row.sledrun_json_rev_id == response['oldrevid'] print(f"Successfully add stätte_id {properties['STAETTE_ID']} to {properties['STAETTE_NA']}") # update wrsledrunjsoncache (not strictly necessary but why not as we have the data) connection.execute(text(""" UPDATE wrsledrunjsoncache SET sledrun_json=:sledrun_json, sledrun_json_rev_id=:sledrun_json_rev_id WHERE sledrun_page_id=:sledrun_page_id AND sledrun_json_page_id=:sledrun_json_page_id; """), [{ 'sledrun_json': new_sledrun_json, 'sledrun_json_rev_id': response['newrevid'], 'sledrun_page_id': row.sledrun_page_id, 'sledrun_json_page_id': row.sledrun_json_page_id, }]) else: print(f'{len(rows)} rows match the name {properties["STAETTE_NA"]}.') else: print(f"{properties['STAETTE_NA']} -> {properties['STAETTE_ID']} (match by stätte_id)") def update_gütesiegel(inifile: List[str], shapefile: str): config = configparser.ConfigParser() config.read(inifile) api_url = config.get('robot', 'wikiurl') api_user = config.get('robot', 'botpassword_bot') api_password = config.get('robot', 'botpassword_password') session = requests.Session() # get login_token get_data = { 'action': 'query', 'meta': 'tokens', 'type': 'login', 'format': 'json', } response = session.get(url=api_url, params=get_data).json() login_token = response['query']['tokens']['logintoken'] # login post_data = { 'action': 'login', 'lgname': api_user, 'lgpassword': api_password, 'lgtoken': login_token, 'format': 'json' } response = session.post(url=api_url, data=post_data).json()['login'] if response['result'] != 'Success': raise RuntimeError(response['reason']) # get csrf_token get_data = { 'action': 'query', 'meta': 'tokens', 'type': 'csrf', 'format': 'json', } response = session.get(url=api_url, params=get_data).json() csrf_token = response['query']['tokens']['csrftoken'] host = config.get('mysql', 'host') dbname = config.get('mysql', 'dbname') user = config.get('mysql', 'user_name') passwd = config.get('mysql', 'user_pass') engine = create_engine(f'mysql://{user}@{host}:3306/{dbname}?passwd={passwd}&charset=utf8mb4') connection = engine.connect() with fiona.open(shapefile) as shp: for feature in shp: update_sledrun(connection, api_url, session, csrf_token, shp.crs, 'EPSG:4326', feature) def main(): parser = argparse.ArgumentParser(description='Update Tiroler Gütesiegel in sledrun JSON ' 'based on official information.') parser.add_argument('inifile', nargs='+', help='inifile.ini, see: https://www.winterrodeln.org/trac/wiki/ConfigIni') parser.add_argument('shapefile', help='Shape file as provided by ' 'https://www.data.gv.at/katalog/dataset/land-tirol_rodelbahnenintirol') args = parser.parse_args() update_gütesiegel(args.inifile, args.shapefile) if __name__ == '__main__': main()