"""Creates an Atom-Feed for single or multiple winterrodeln sled reports. Format: https://www.winterrodeln.org/feed/berichte/alle https://www.winterrodeln.org/feed/berichte/bahn/kemater_alm https://www.winterrodeln.org/feed/berichte/bahnen/5+280+251 https://www.winterrodeln.org/feed/berichte/region/osttirol See: https://www.winterrodeln.org/trac/wiki/UrlSchema """ import datetime from xml.etree.ElementTree import Element, SubElement, tostring from flask import url_for from sqlalchemy import func, desc class CreateFeedError(RuntimeError): pass class RegionNotFoundError(CreateFeedError): pass def create_feed(db, limit, page_title=None, page_ids=None, region_name=None): """If a page_title is given, only the reports for the given sledrun are shown. If a list of page_ids is given, only the reports for the selected pages are shown. If a region name (lower case) is given, the reports just for this region are shown. Otherwise, all reports are shown.""" wrreport = db.metadata.tables['wrreport'] wrsledruncache = db.metadata.tables['wrsledruncache'] wrregion = db.metadata.tables['wrregion'] wrregioncache = db.metadata.tables['wrregioncache'] sql = db.select([ wrreport, wrsledruncache.c.position_longitude.label('lon'), wrsledruncache.c.position_latitude.label('lat') ]). \ select_from(wrreport.outerjoin(wrsledruncache, wrreport.c.page_id == wrsledruncache.c.page_id)). \ where(wrreport.c.date_invalid > func.now()). \ where(wrreport.c.delete_date.is_(None)). \ order_by(desc(wrreport.c.id)). \ limit(limit) if page_title is not None: # page_title is given page_title = page_title.replace('_', ' ') sql = sql.where(func.lcase(wrreport.c.page_title) == func.lcase(page_title)) elif page_ids is not None: # a list of page_ids is given sql = sql.where(wrreport.c.page_id.in_(page_ids)) elif region_name is not None: # a name of a region is given # (1) find out whether the region exists sub_sql = db.select([wrregion]).where(wrregion.c.name == func.lcase(region_name)) sub_result = db.session.connection().execute(sub_sql) if sub_result.rowcount == 0: # no region with such a name raise RegionNotFoundError(region_name) assert sub_result.rowcount == 1 sub_row = sub_result.fetchone() region_id = sub_row.id # (2) now we have the id of the region sql = sql.where(wrreport.c.page_id.in_(db.select([wrregioncache.c.page_id]) .where(region_id == wrregioncache.c.region_id))) else: # user wants to have all reports pass result = db.session.connection().execute(sql) feed = Element("feed", xmlns="http://www.w3.org/2005/Atom", attrib={'xmlns:georss': 'http://www.georss.org/georss', 'xmlns:wr': 'http://www.winterrodeln.org/schema/wrreport'}) feed_title = SubElement(feed, "title") feed_title.text = "Winterrodeln Rodelbahnberichte" feed_id = SubElement(feed, "id") if page_title is not None: feed_id.text = url_for('bahn', bahn=page_title, _external=True) elif page_ids is not None: feed_id.text = url_for('bahnen', bahnen="+".join(map(str, page_ids)), _external=True) elif region_name is not None: feed_id.text = url_for('region', region=region_name, _external=True) else: feed_id.text = url_for('alle', _external=True) feed_updated = SubElement(feed, "updated") feed.append(Element("link", rel="self", href=feed_id.text)) last_updated = None for row in result: page_title_url = row.page_title.replace(' ', '_') entry = SubElement(feed, "entry") entry_title = SubElement(entry, "title") entry_title.text = row.page_title entry.append(Element("link", rel="alternate", href="https://www.winterrodeln.org/wiki/{0}".format(page_title_url), type="text/html", hreflang="de")) entry_id = SubElement(entry, "id") entry_id.text = "https://www.winterrodeln.org/wiki/{0}#{1}".format(page_title_url, row.id) entry_updated = SubElement(entry, "updated") entry_updated.text = row.date_entry.isoformat() + "+01:00" if last_updated is None: last_updated = row.date_entry # entry_summary = SubElement(entry, "summary") # entry_summary.text = str(condition) entry_content = SubElement(entry, "content") entry_content.attrib["type"] = "xhtml" entry_content_div = SubElement(entry_content, "div") entry_content_div.attrib["xmlns"] = "http://www.w3.org/1999/xhtml" entry_content_ul = SubElement(entry_content_div, "ul") if row.date_report is not None: entry_content_date = SubElement(entry_content_ul, "li") entry_content_date.text = "Bericht für " + row.date_report.isoformat() if row.condition is not None: entry_content_condition = SubElement(entry_content_ul, "li") entry_content_condition.text = "Schneelage: " + \ {1: 'Sehr gut', 2: 'Gut', 3: 'Mittelmäßig', 4: 'Schlecht', 5: 'Geht nicht'}[row.condition] entry_content_description = SubElement(entry_content_ul, "li") entry_content_description.text = row.description entry_author = SubElement(entry, "author") entry_author_name = SubElement(entry_author, "name") if row.author_name is None or len(row.author_name.strip()) == 0: entry_author_name.text = "Anonymous user" else: entry_author_name.text = row.author_name if row.lon is not None and row.lat is not None: entry_geo = SubElement(entry, "georss:point") entry_geo.text = "{lat} {lon}".format(lat=row.lat, lon=row.lon) entry_wrreport = SubElement(entry, "wr:report") entry_wrreport.attrib['report_id'] = str(row.id) entry_wrreport.attrib['page_id'] = str(row.page_id) entry_wrreport.attrib['page_title'] = row.page_title entry_wrreport.attrib['date_report'] = str(row.date_report) entry_wrreport.attrib['date_entry'] = str(row.date_entry) entry_wrreport.attrib['condition'] = "0" if row.condition is None else str(row.condition) entry_wrreport.attrib['author_name'] = row.author_name entry_wrreport.attrib['author_username'] = "" if row.author_userid is None else row.author_username entry_wrreport.text = row.description if last_updated is None: last_updated = datetime.datetime.now() feed_updated.text = last_updated.isoformat() + "+01:00" return b'\n' + tostring(feed)