reimplemented openMensaFeedv2 and fixed linting

This commit is contained in:
Hadrian Burkhardt
2023-12-03 02:51:47 +01:00
committed by f4lco
parent 9e37940334
commit d637d71c8f
16 changed files with 632 additions and 598 deletions
-412
View File
@@ -1,412 +0,0 @@
from xml.dom import minidom
from datetime import date
class Builder:
"""A class method for creating a new class."""
def __init__(self):
"""Initialize the object for the OpenMensa Feed Doc XML."""
self._doc = minidom.Document()
self._om = self._doc.createElement("openmensa")
self._om.setAttribute("version", "2.1")
self._om.setAttribute("xmlns", "http://openmensa.org/open-mensa-v2")
self._om.setAttribute(
"xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance"
)
self._om.setAttribute(
"xsi:schemaLocation",
"http://openmensa.org/open-mensa-v2 "
+ "http://openmensa.org/open-mensa-v2.xsd",
)
self._version = None
self._name = None
self._address = None
self._city = None
self._phone = None
self._email = None
self._location = None
self._availability = None
self._times = None
self._feed = None
self._day = None
self._days: dict[date, dict[str, list]] = {}
@property
def version(self):
"""The version of the device .
Returns:
[type]: [description]
"""
return self._version
@version.setter
def version(self, value: str):
self._version = value
@version.deleter
def version(self):
del self._version
@property
def name(self):
"""Name of the canteen .
Returns:
[type]: [description]
"""
return self._name
@name.setter
def name(self, value: str):
self._name = value
@name.deleter
def name(self):
del self._name
@property
def address(self):
"""The address of the canteen .
Returns:
[type]: [description]
"""
return self._address
@address.setter
def address(self, value: tuple[str, str, str]):
street_nr, zip_code, city = value
self._address = f"{street_nr}, {zip_code} {city}"
@address.deleter
def address(self):
del self._address
@property
def city(self):
"""Get the city of the canteen .
Returns:
[type]: [description]
"""
return self._city
@city.setter
def city(self, value: str):
self._city = value
@city.deleter
def city(self):
del self._city
@property
def phone(self):
"""The phone number .
Returns:
[type]: [description]
"""
return self._phone
@phone.setter
def phone(self, value: str):
self._phone = value
@phone.deleter
def phone(self):
del self._phone
@property
def email(self):
"""The email address of the canteen .
Returns:
[type]: [description]
"""
return self._email
@email.setter
def email(self, value: str):
self._email = value
@email.deleter
def email(self):
del self._email
@property
def location(self):
"""Get a tuple containing the location as latitude and longitude .
Returns:
[type]: [description]
"""
return (self._longitude, self._latitude)
@location.setter
def location(self, value: tuple[float, float]):
self._longitude = value[0]
self._latitude = value[1]
@location.deleter
def location(self):
del self._longitude
del self._latitude
@property
def availability(self):
"""Whether the canteen is public or restriced.
Returns:
[type]: [description]
"""
return self._availability
@availability.setter
def availability(self, value: str):
if value == "pulbic" or value == "restricted":
self._availability = value
else:
raise ValueError("only 'public' or 'restricted are allowed.")
@availability.deleter
def availability(self):
del self._availability
@property
def times(self):
"""Get the opening times the canteen.
Returns:
[type]: [description]
"""
return self._times
@times.setter
def times(self, value: dict[str, str]):
def attach_weekday(tag: str, value: str):
"""Attach a week tag to the week .
Args:
tag (str): [description]
value (str): [description]
"""
if value == "":
return
d = self._doc.createElement(tag)
if value == "geschlossen":
d.setAttribute("closed", "true")
else:
d.setAttribute("open", value)
self._times.appendChild(d)
weekdays = (
"monday",
"tuesday",
"wednesday",
"thursday",
"friday",
"saturday",
"sunday",
)
self._times = self._doc.createElement("times")
self._times.setAttribute("type", "opening")
for weekday in weekdays:
v = value.get(weekday)
if v:
attach_weekday(weekday, v)
@times.deleter
def times(self):
del self._times
@property
def feed(self):
"""Get a feed object .
Returns:
[type]: [description]
"""
return self._feed
@feed.setter
def feed(self, value: dict):
"""Set the feed element .
Args:
value (dict): [description]
"""
name: str = value.get("name")
priority: int = value.get("priority")
url: str = value.get("url")
source: str = value.get("source")
hour: int = value.get("hour")
dayOfMonth: int | str = (
value.get("dayOfMonth") if value.get("dayOfMonth") else "*"
)
dayOfWeek: int | str = (
value.get("dayOfWeek") if value.get("dayOfWeek") else "*"
)
month: int | str = value.get("month") if value.get("month") else "*"
minute: int = value.get("minute") if value.get("minute") else 0
retry: str = value.get("retry")
self._feed = self._doc.createElement("feed")
self._feed.setAttribute("name", name)
self._feed.setAttribute("priority", str(priority))
schedule = self._doc.createElement("schedule")
self._feed.appendChild(schedule)
schedule.setAttribute("dayOfMonth", str(dayOfMonth))
schedule.setAttribute("dayOfWeek", str(dayOfWeek))
schedule.setAttribute("month", str(month))
schedule.setAttribute("hour", str(hour))
schedule.setAttribute("minute", str(minute))
if retry:
schedule.setAttribute("retry", retry)
el = self._doc.createElement("url")
self._feed.appendChild(el)
node = self._doc.createTextNode(url)
el.appendChild(node)
el = self._doc.createElement("source")
self._feed.appendChild(el)
node = self._doc.createTextNode(source)
el.appendChild(node)
@feed.deleter
def feed(self):
del self._feed
@property
def day(self):
"""Returns the number of day of the week .
Returns:
[type]: [description]
"""
return self._day
@day.setter
def day(self, value):
self._day = value
@day.deleter
def day(self):
del self._day
def add_meal(
self,
date: date,
category: str,
name: str,
prices: dict[str, float],
note: str = None,
):
"""Add a meal element to the document .
Args:
date (date): [description]
category (str): [description]
name (str): [description]
prices (dict[str, float]): [description]
note (str, optional): [description]. Defaults to None.
"""
meal = self._doc.createElement("meal")
node = self._doc.createElement("name")
meal.appendChild(node)
data = self._doc.createTextNode(name)
node.appendChild(data)
if note:
node = self._doc.createElement("note")
meal.appendChild(node)
data = self._doc.createTextNode(note)
node.appendChild(data)
if prices["student"]:
node = self._doc.createElement("price")
meal.appendChild(node)
node.setAttribute("role", "student")
data = self._doc.createTextNode(f'{prices["student"]:.2f}')
node.appendChild(data)
if prices["employee"]:
node = self._doc.createElement("price")
meal.appendChild(node)
node.setAttribute("role", "employee")
data = self._doc.createTextNode(f'{prices["employee"]:.2f}')
node.appendChild(data)
if prices["other"]:
node = self._doc.createElement("price")
meal.appendChild(node)
node.setAttribute("role", "other")
data = self._doc.createTextNode(f'{prices["other"]:.2f}')
node.appendChild(data)
category_dict = self._days.get(date, dict())
if not category_dict:
self._days[date] = category_dict
meal_list = category_dict.get(category, list())
if not meal_list:
category_dict[category] = meal_list
meal_list.append(meal)
def __append_node(self, tag: str, value: str):
"""Create a node with a tag and text .
Args:
tag (str): [description]
value (str): [description]
"""
elem = self._doc.createElement(tag)
self._canteen.appendChild(elem)
node = self._doc.createTextNode(value)
elem.appendChild(node)
def toXML(self):
"""Return a XML string representing the canteen.
Returns:
[type]: [description]
"""
self._doc.appendChild(self._om)
if self.version:
self.__append_node("version", self.version)
self._canteen = self._doc.createElement("canteen")
self._om.appendChild(self._canteen)
if self.name:
self.__append_node("name", self.name)
if self.address:
self.__append_node("address", self.address)
if self.city:
self.__append_node("city", self.city)
if self.phone:
self.__append_node("phone", self.phone)
if self.email:
self.__append_node("email", self.email)
if self._longitude and self._latitude:
location = self._doc.createElement("location")
self._canteen.appendChild(location)
location.setAttribute("latitude", str(self._latitude))
location.setAttribute("longitude", str(self._longitude))
if self.availability:
self.__append_node("availability", self.availability)
if self.times:
self._canteen.appendChild(self.times)
if self.feed:
self._canteen.appendChild(self.feed)
for date, category_dict in sorted(self._days.items()):
day = self._doc.createElement("day")
self._canteen.appendChild(day)
day.setAttribute("date", str(date))
for category_name, meals in sorted(category_dict.items()):
category = self._doc.createElement("category")
day.appendChild(category)
category.setAttribute("name", category_name)
for meal in meals:
category.appendChild(meal)
return self._doc.toprettyxml(encoding="UTF-8")
+1 -1
View File
@@ -34,7 +34,7 @@ id = 356
cHash = 58cfcf13b92d8045c0810bcca34c37e7 cHash = 58cfcf13b92d8045c0810bcca34c37e7
[brandenburg] [brandenburg]
name = Mensa Brandenburg an der Havel name = Mensa Brandenburg
street = Magdeburger Straße 50 street = Magdeburger Straße 50
city = 14770 Brandenburg an der Havel city = 14770 Brandenburg an der Havel
id = 357 id = 357
+75 -45
View File
@@ -5,7 +5,7 @@ import time
import json import json
class SWP_Webspeiseplan_API: class SWPWebspeiseplanAPI:
"""This class is used download content from SWP_Webspeiseplan. """This class is used download content from SWP_Webspeiseplan.
Returns: Returns:
@@ -13,53 +13,24 @@ class SWP_Webspeiseplan_API:
""" """
URL_BASE = "https://swp.webspeiseplan.de" URL_BASE = "https://swp.webspeiseplan.de"
logger = logging.getLogger(__name__)
def __init__(self): def __init__(self):
"""Initialize the configuration for the web service .""" """Initialize the configuration for the web service."""
logging.basicConfig() logging.basicConfig()
self.logger = logging.getLogger(__name__) proxy_token = self.parse_token()
self.__parse_token() self.outlets = self.parse_outlets(proxy_token)
params = { self.menus: dict[str, dict] = {}
"token": self.proxy_token, self.meal_categories: dict[str, dict] = {}
"model": "outlet",
"location": "",
"languagetype": "",
"_": int(time.time() * 1000),
}
self.outlets = {
outlet["name"]: outlet for outlet in self.__parse_model(params)
}
self.menus = {}
self.meal_categories = {}
for outlet in self.outlets.values(): for outlet in self.outlets.values():
params["model"] = "menu" location = outlet["standortID"]
params["location"] = outlet["standortID"] menu = self.parse_menu(proxy_token, location)
params["languagetype"] = 1 categories = self.parse_meal_category(proxy_token, location)
params["_"] = int(time.time() * 1000)
menu = self.__parse_model(params)
self.menus[outlet["name"]] = menu
params["model"] = "mealCategory"
params["_"] = int(time.time() * 1000)
categories = self.__parse_model(params)
id2cat = {item["gerichtkategorieID"]: item for item in categories} id2cat = {item["gerichtkategorieID"]: item for item in categories}
self.menus[outlet["name"]] = menu
self.meal_categories[outlet["name"]] = id2cat self.meal_categories[outlet["name"]] = id2cat
def __parse_token(self): def __spoof_req_headers(self, req: urllib.request.Request):
"""Get the token from the proxy server."""
req = urllib.request.Request(self.URL_BASE)
with urllib.request.urlopen(req) as resp:
txt = resp.read().decode("utf-8")
match = re.findall(r"/main.[0-9a-f]+.js", txt)[0]
self.logger.debug(f"__parse_token: downloading script {match}")
req = urllib.request.Request(f"{self.URL_BASE}{match}")
with urllib.request.urlopen(req) as resp:
txt = resp.read().decode("utf-8")
self.proxy_token = re.findall(r"PROXY_TOKEN:\"([0-9a-f]+)\"", txt)[0]
self.logger.debug(f"__parse_token: PROXY_TOKEN {self.proxy_token}")
def __spoof_req_headers(req: urllib.request.Request):
"""Add headers to a request . """Add headers to a request .
Args: Args:
@@ -91,7 +62,7 @@ class SWP_Webspeiseplan_API:
) )
req.add_header("X-Requested-With", "XMLHttpRequest") req.add_header("X-Requested-With", "XMLHttpRequest")
def __parse_model(self, params: dict): def parse_model(self, params: dict):
"""Retrieve data from host. """Retrieve data from host.
Args: Args:
@@ -100,12 +71,71 @@ class SWP_Webspeiseplan_API:
Returns: Returns:
[type]: [description] [type]: [description]
""" """
url = f"{self.URL_BASE}/index.php?" + "&".join( url = f"{SWPWebspeiseplanAPI.URL_BASE}/index.php?" + "&".join(
[f"{k}={v}" for k, v in params.items()] [f"{k}={v}" for k, v in params.items()]
) )
self.logger.debug(f"__parse_model: {url}") SWPWebspeiseplanAPI.logger.debug("__parse_model: %s", url)
req = urllib.request.Request(url) req = urllib.request.Request(url)
SWP_Webspeiseplan_API.__spoof_req_headers(req) self.__spoof_req_headers(req)
with urllib.request.urlopen(req) as resp: with urllib.request.urlopen(req) as resp:
data = resp.read() data = resp.read()
return json.loads(data)["content"] return json.loads(data)["content"]
def parse_token(self) -> str:
"""Get the token from the proxy server."""
req = urllib.request.Request(SWPWebspeiseplanAPI.URL_BASE)
with urllib.request.urlopen(req) as resp:
txt = resp.read().decode("utf-8")
match = re.findall(r"/main.[0-9a-f]+.js", txt)[0]
SWPWebspeiseplanAPI.logger.debug(
"__parse_token: downloading script %s", match
)
req = urllib.request.Request(f"{SWPWebspeiseplanAPI.URL_BASE}{match}")
with urllib.request.urlopen(req) as resp:
txt = resp.read().decode("utf-8")
proxy_token = re.findall(r"PROXY_TOKEN:\"([0-9a-f]+)\"", txt)[0]
SWPWebspeiseplanAPI.logger.debug(
"__parse_token: PROXY_TOKEN %s", proxy_token
)
return proxy_token
def parse_outlets(self, proxy_token: str) -> dict[str, dict]:
"""Get the outlets from the server."""
params = {
"token": proxy_token,
"model": "outlet",
"location": "",
"languagetype": "",
"_": int(time.time() * 1000),
}
outlets = {
outlet["name"]: outlet for outlet in self.parse_model(params)
}
return outlets
def parse_menu(self, proxy_token: str, location: int) -> dict:
"""Get the menu for a specific location."""
params = {
"token": proxy_token,
"model": "menu",
"location": location,
"languagetype": 1,
"_": int(time.time() * 1000),
}
menu = self.parse_model(params)
return menu
def parse_meal_category(
self, proxy_token: str, location: int
) -> list[dict]:
"""Get the meal catrgories for a specific location."""
params = {
"token": proxy_token,
"model": "mealCategory",
"location": location,
"languagetype": 1,
"_": int(time.time() * 1000),
}
menu = self.parse_model(params)
return menu
+44 -80
View File
@@ -1,63 +1,43 @@
import logging import logging
from datetime import datetime from datetime import datetime, date
from stw_potsdam.builder import Builder from stw_potsdam.xml_types.canteen_xml import CanteenMeta, CanteenXML
from stw_potsdam.swp_webspeiseplan_api import SWP_Webspeiseplan_API from stw_potsdam.xml_types.times_xml import TimesXML
from stw_potsdam.xml_types.meal_xml import MealXML
class SWP_Webspeiseplan_Parser: class SWPWebspeiseplanParser:
"""Class method to parse SWP_Webspeiseplan.""" """Class method to parse SWP_Webspeiseplan."""
def __init__( def __init__(self) -> None:
self, """Init SWPWebspeiseplanParser object."""
menu_data: list[dict],
meal_categories: list[dict],
outlet_data: dict,
url: str,
):
"""Initialize the parser .
Args:
menu_data (list[dict]): [description]
meal_categories (list[dict]): [description]
outlet_data (dict): [description]
url (str): [description]
"""
logging.basicConfig() logging.basicConfig()
self.logger = logging.getLogger(__name__) self.logger = logging.getLogger(__name__)
self.menu_data = menu_data
self.meal_categories = meal_categories
self.outlet_data = outlet_data
self.url = url
self._builder = Builder()
self.__parse_canteen(outlet_data)
self.__parse_feed()
self.__parse_meals()
def __parse_canteen(self, outlet: dict): def parse_canteen_meta_times(self, outlet: dict):
"""Parse the outlet data from outlet. """Parse the outlet data from outlet.
Args: Args:
outlet (dict): [description] outlet (dict): [description]
""" """
canteen = self._builder self.logger.debug("parse_canteen_meta_times(): %s", outlet["name"])
canteen.name = outlet["name"] addr_info = outlet["addressInfo"]
canteen.address = ( meta = {
outlet["addressInfo"]["street"], "name": outlet["name"],
outlet["addressInfo"]["postalCode"], "address": f'{addr_info["street"]}, {addr_info["postalCode"]} '
outlet["addressInfo"]["city"], + f'{addr_info["city"]}',
) "city": addr_info["city"],
canteen.city = outlet["addressInfo"]["city"] "phone": outlet["contactInfo"][0]["phone"],
canteen.phone = outlet["contactInfo"][0]["phone"] "email": outlet["contactInfo"][0]["email"],
canteen.email = outlet["contactInfo"][0]["email"] }
if outlet["positionInfo"]: if outlet["positionInfo"]:
canteen.location = ( meta["location"] = (
outlet["positionInfo"]["longitude"], outlet["positionInfo"]["longitude"],
outlet["positionInfo"]["latitude"], outlet["positionInfo"]["latitude"],
) )
canteen_meta = CanteenMeta(**meta)
# TODO: availability via locations isPublic # TODO: availability via locations isPublic
weekday_dict = {
times = {
"monday": f"{outlet['moZeit1']}, {outlet['moZeit2']}", "monday": f"{outlet['moZeit1']}, {outlet['moZeit2']}",
"tuesday": f"{outlet['diZeit1']}, {outlet['diZeit2']}", "tuesday": f"{outlet['diZeit1']}, {outlet['diZeit2']}",
"wednesday": f"{outlet['miZeit1']}, {outlet['miZeit2']}", "wednesday": f"{outlet['miZeit1']}, {outlet['miZeit2']}",
@@ -67,52 +47,36 @@ class SWP_Webspeiseplan_Parser:
"sunday": f"{outlet['soZeit1']}, {outlet['soZeit2']}", "sunday": f"{outlet['soZeit1']}, {outlet['soZeit2']}",
} }
times = { weekday_dict = {
k: v.replace("None, None", "") k: v.replace("None, None", "")
.replace("None,", "") .replace("None,", "")
.replace(", None", "") .replace(", None", "")
for k, v in times.items() for k, v in weekday_dict.items()
} }
canteen.times = times canteen_times = TimesXML(weekday_dict)
canteen = CanteenXML(canteen_meta, canteen_times)
return canteen
def __parse_feed(self): def parse_meals(
"""Parse feed and set feed.""" self, menu_data, meal_categories
feed = { ) -> list[tuple[date, str, MealXML]]:
"name": "full",
"priority": 0,
"hour": "8-14",
"retry": "30 1",
"url": self.url,
"source": SWP_Webspeiseplan_API.URL_BASE,
}
self._builder.feed = feed
def __parse_meals(self):
"""Parse the menu and adds it to the builder.""" """Parse the menu and adds it to the builder."""
for menu in self.menu_data: meals = []
for meal in menu["speiseplanGerichtData"]: for menu in menu_data:
info = meal["speiseplanAdvancedGericht"] for meal_data in menu["speiseplanGerichtData"]:
date = datetime.fromisoformat(info["datum"]).date() info = meal_data["speiseplanAdvancedGericht"]
additional_info = meal["zusatzinformationen"] additional_info = meal_data["zusatzinformationen"]
self._builder.add_meal( price = {
date=date,
category=self.meal_categories[info["gerichtkategorieID"]][
"name"
],
name=info["gerichtname"],
prices={
"student": additional_info["mitarbeiterpreisDecimal2"], "student": additional_info["mitarbeiterpreisDecimal2"],
"employee": additional_info["price3Decimal2"], "employee": additional_info["price3Decimal2"],
"other": additional_info["gaestepreisDecimal2"], "other": additional_info["gaestepreisDecimal2"],
}, }
meal = MealXML(name=info["gerichtname"], price=price)
day = datetime.fromisoformat(info["datum"]).date()
category = meal_categories[info["gerichtkategorieID"]]["name"]
meals.append(
{"day": day, "category": category, "meal": meal}
) )
self.logger.debug("parse_meals(): %s meals parsed", len(meals))
@property return meals
def xml_feed(self):
"""Return the XML string of the builder.
Returns:
[type]: [description]
"""
return self._builder.toXML()
+9 -21
View File
@@ -9,8 +9,7 @@ from flask import Flask, jsonify, make_response, url_for
from flask.logging import create_logger from flask.logging import create_logger
from stw_potsdam.config import read_canteen_config from stw_potsdam.config import read_canteen_config
from stw_potsdam.swp_webspeiseplan_api import SWP_Webspeiseplan_API from stw_potsdam.xml_types.builder import Builder
from stw_potsdam.swp_webspeiseplan_parser import SWP_Webspeiseplan_Parser
CACHE_TIMEOUT = 45 * 60 CACHE_TIMEOUT = 45 * 60
@@ -18,7 +17,8 @@ CACHE_TIMEOUT = 45 * 60
app = Flask(__name__) app = Flask(__name__)
app.url_map.strict_slashes = False app.url_map.strict_slashes = False
cache = ct.TTLCache(maxsize=30, ttl=CACHE_TIMEOUT)
config = read_canteen_config()
log = create_logger(app) log = create_logger(app)
if "BASE_URL" in os.environ: # pragma: no cover if "BASE_URL" in os.environ: # pragma: no cover
@@ -30,10 +30,8 @@ if "BASE_URL" in os.environ: # pragma: no cover
if base_url.path: if base_url.path:
app.config["APPLICATION_ROOT"] = base_url.path app.config["APPLICATION_ROOT"] = base_url.path
cache = ct.TTLCache(maxsize=30, ttl=CACHE_TIMEOUT)
def canteen_not_found(canteen_name):
def canteen_not_found(config, canteen_name):
log.warning("Canteen %s not found", canteen_name) log.warning("Canteen %s not found", canteen_name)
configured = ", ".join(f"'{c}'" for c in config.keys()) configured = ", ".join(f"'{c}'" for c in config.keys())
message = f"Canteen '{canteen_name}' not found, available: {configured}" message = f"Canteen '{canteen_name}' not found, available: {configured}"
@@ -41,28 +39,19 @@ def canteen_not_found(config, canteen_name):
@ct.cached(cache=cache) @ct.cached(cache=cache)
def get_menu(): def update_builder():
log.debug("Downloading menu for SWP") log.debug("Downloading menu for SWP")
return SWP_Webspeiseplan_API() return Builder(config)
@app.route("/canteens/<canteen_name>") @app.route("/canteens/<canteen_name>")
@app.route("/canteens/<canteen_name>/xml") @app.route("/canteens/<canteen_name>/xml")
def canteen_xml_feed(canteen_name): def canteen_xml_feed(canteen_name):
config = read_canteen_config()
if canteen_name not in config: if canteen_name not in config:
return canteen_not_found(config, canteen_name) return canteen_not_found(canteen_name)
canteen = config[canteen_name] builder = update_builder()
swp_api = get_menu() xml = builder.get_xml(canteen_name)
swp_parser = SWP_Webspeiseplan_Parser(
swp_api.menus[canteen.name],
swp_api.meal_categories[canteen.name],
swp_api.outlets[canteen.name],
url_for("canteen_xml_feed", canteen_name=canteen.key, _external=True),
)
xml = swp_parser.xml_feed.decode()
response = make_response(xml) response = make_response(xml)
response.mimetype = "text/xml" response.mimetype = "text/xml"
return response return response
@@ -71,7 +60,6 @@ def canteen_xml_feed(canteen_name):
@app.route("/") @app.route("/")
@app.route("/canteens") @app.route("/canteens")
def canteen_index(): def canteen_index():
config = read_canteen_config()
return jsonify( return jsonify(
{ {
key: url_for("canteen_xml_feed", canteen_name=key, _external=True) key: url_for("canteen_xml_feed", canteen_name=key, _external=True)
View File
+67
View File
@@ -0,0 +1,67 @@
from xml.dom import minidom
from dataclasses import dataclass
import logging
from flask import url_for
from stw_potsdam.xml_types.openmensa_xml import OpenMensaXML
from stw_potsdam.swp_webspeiseplan_api import SWPWebspeiseplanAPI
from stw_potsdam.swp_webspeiseplan_parser import SWPWebspeiseplanParser
from stw_potsdam.config import Canteen
from stw_potsdam.xml_types.feed_xml import FeedXML, ScheduleXML
@dataclass
class Builder:
"""A class method for creating a new OpenMensa Feed."""
VERSION = "2.0.1"
def __init__(self, config: dict[str, Canteen]):
"""Initialize the object for the OpenMensa Feed Doc XML."""
logging.basicConfig()
self.logger = logging.getLogger(__name__)
self._xml_data = {}
swp_api = SWPWebspeiseplanAPI()
swp_parser = SWPWebspeiseplanParser()
for cname, ntup in config.items():
if ntup.name not in swp_api.outlets.keys():
self.logger.warning("%s not found in keys", ntup.name)
continue
outlet = swp_api.outlets[ntup.name]
menus = swp_api.menus[ntup.name]
categories = swp_api.meal_categories[ntup.name]
canteen = swp_parser.parse_canteen_meta_times(outlet)
meals = swp_parser.parse_meals(menus, categories)
for kwargs in meals:
canteen.add_meal(**kwargs)
feed = self.__create_feed(ntup)
canteen.add_feed(feed)
self._xml_data[cname] = OpenMensaXML(self.VERSION, canteen)
def __create_feed(self, ntup: Canteen):
schedule = ScheduleXML(
hour="8-14",
retry="30 1",
)
feed = FeedXML(
name="full",
priority=0,
source=SWPWebspeiseplanAPI.URL_BASE,
url=url_for(
"canteen_xml_feed",
canteen_name=ntup.key,
_external=True,
),
schedule=schedule,
)
return feed
def get_xml(self, canteen_name: str):
"""Return a XML string representing the canteen.
Returns:
[type]: [description]
"""
doc = minidom.Document()
xml_element = self._xml_data[canteen_name].xml_element(doc)
doc.appendChild(xml_element)
return doc.toprettyxml(encoding="UTF-8")
+151
View File
@@ -0,0 +1,151 @@
from dataclasses import dataclass
from xml.dom import minidom
from datetime import date
from stw_potsdam.xml_types.times_xml import TimesXML
from stw_potsdam.xml_types.meal_xml import MealXML
from stw_potsdam.xml_types.feed_xml import FeedXML
@dataclass
class CanteenMeta:
"""Metadata for CanteenXML."""
# pylint: disable=too-many-instance-attributes
def __init__(self, **kwargs):
"""Init CanteenMeta object."""
self.name: str = kwargs["name"]
self.address: str = kwargs["address"]
self.city: str = kwargs["city"]
self.phone: str = kwargs["phone"]
self.email = kwargs["email"]
self.location: tuple[float, float] = kwargs.get("location", None)
self.availability: str = kwargs.get("availability", "public")
@property
def availability(self) -> str:
"""Whether the canteen is public or restricted.
Returns:
str: 'public' | 'restricted'
"""
return self._availability
@availability.setter
def availability(self, value: str):
if value in ("public", "restricted"):
self._availability = value
else:
raise ValueError("only 'public' or 'restricted' are allowed.")
@availability.deleter
def availability(self):
del self._availability
@dataclass
class CanteenXML:
"""Represents the canteen tag in openMensaFeedv2."""
def __init__(
self,
canteen_meta: CanteenMeta,
times: TimesXML,
feeds: dict[str, FeedXML] = None,
days: dict[date, dict[str, list[MealXML]]] = None,
):
"""Init CanteenXML Object.
Args:
name (str): _description_
address (str): _description_
city (str): _description_
phone (str): _description_
email (str): _description_
location (tuple[float, float]): _description_
availability (str): _description_
times (TimesXML): _description_
"""
self.canteen_meta = canteen_meta
self.times = times
self.feeds = {} if feeds is None else feeds
self.days = {} if days is None else days
def __create_node(self, doc: minidom.Document, tag: str, value: str):
e = doc.createElement(tag)
tn = doc.createTextNode(value)
e.appendChild(tn)
return e
def __append_meta(self, doc: minidom.Document, canteen: minidom.Element):
name = self.__create_node(doc, "name", self.canteen_meta.name)
canteen.appendChild(name)
address = self.__create_node(doc, "address", self.canteen_meta.address)
canteen.appendChild(address)
city = self.__create_node(doc, "city", self.canteen_meta.city)
canteen.appendChild(city)
phone = self.__create_node(doc, "phone", self.canteen_meta.phone)
canteen.appendChild(phone)
email = self.__create_node(doc, "email", self.canteen_meta.email)
canteen.appendChild(email)
if self.canteen_meta.location:
location = doc.createElement("location")
location.setAttribute("longitude", str(self.canteen_meta.location[0]))
location.setAttribute("latitude", str(self.canteen_meta.location[1]))
canteen.appendChild(location)
availability = self.__create_node(
doc, "availability", self.canteen_meta.availability
)
canteen.appendChild(availability)
times = self.times.xml_element(doc)
canteen.appendChild(times)
def add_feed(self, feed: FeedXML):
"""Add a feed to the canteen.
Args:
feed (FeedXML): _description_
"""
self.feeds[feed.name] = feed
def add_meal(self, day: date, category: str, meal: MealXML):
"""Add a meal to the canteen.
Args:
day (date): Offered date of meal.
catrgory (str): Meal's category.
meal (MealXML): The meal item.
"""
categories = self.days.get(day, {})
if not categories:
self.days[day] = categories
meals = categories.get(category, [])
if not meals:
categories[category] = meals
meals.append(meal)
def xml_element(self, doc: minidom.Document):
"""Return the XML representation.
Args:
doc (minidom.Document): Working XML document
Returns:
_type_: _description_
"""
canteen = doc.createElement("canteen")
self.__append_meta(doc, canteen)
for feed_item in self.feeds.values():
feed = feed_item.xml_element(doc)
canteen.appendChild(feed)
for day_data, categories in self.days.items():
day = doc.createElement("day")
day.setAttribute("date", str(day_data))
for category_name, meals in categories.items():
category = doc.createElement("category")
category.setAttribute("name", category_name)
for meal in meals:
category.appendChild(meal.xml_element(doc))
day.appendChild(category)
canteen.appendChild(day)
return canteen
+82
View File
@@ -0,0 +1,82 @@
from dataclasses import dataclass
from xml.dom import minidom
@dataclass
class ScheduleXML:
"""Represents the schedule inside the feed tag in openMensaFeedv2."""
def __init__(self, hour: str, **kwargs):
"""Init ScheduleXML object.
Args:
hour (str): _description_
day_of_month (str, optional): _description_. Defaults to "*".
day_of_week (str, optional): _description_. Defaults to "*".
month (str, optional): _description_. Defaults to "*".
minute (int, optional): _description_. Defaults to 0.
retry (str, optional): _description_. Defaults to None.
"""
self.hour = hour
self.day_of_month = kwargs.get("day_of_month", "*")
self.day_of_week = kwargs.get("day_of_week", "*")
self.month = kwargs.get("month", "*")
self.minute = kwargs.get("minute", 0)
self.retry = kwargs.get("retry", None)
def xml_element(self, doc: minidom.Document):
"""Return the XML representaion.
Args:
doc (minidom.Document): Working XML document.
Returns:
_type_: _description_
"""
schedule = doc.createElement("schedule")
schedule.setAttribute("dayOfMonth", self.day_of_month)
schedule.setAttribute("dayOfWeek", self.day_of_week)
schedule.setAttribute("month", self.month)
schedule.setAttribute("hour", self.hour)
schedule.setAttribute("minute", str(self.minute))
if self.retry:
schedule.setAttribute("retry", self.retry)
return schedule
@dataclass
class FeedXML:
"""Represents the feed tag in openMensaFeedv2."""
name: str
source: str
url: str
schedule: ScheduleXML
priority: int = 0
def xml_element(self, doc: minidom.Document):
"""Return the XML representaion.
Args:
doc (minidom.Document): Working XML document.
Returns:
_type_: _description_
"""
feed = doc.createElement("feed")
feed.setAttribute("name", self.name)
feed.setAttribute("priority", str(self.priority))
schedule = self.schedule.xml_element(doc)
feed.appendChild(schedule)
url = doc.createElement("url")
tn = doc.createTextNode(self.url)
url.appendChild(tn)
feed.appendChild(url)
source = doc.createElement("source")
tn = doc.createTextNode(self.source)
source.appendChild(tn)
feed.appendChild(source)
return feed
+46
View File
@@ -0,0 +1,46 @@
from xml.dom import minidom
from dataclasses import dataclass
@dataclass
class MealXML:
"""Represents the meal tag in openMensaFeedv2."""
def __init__(self, name: str, price: dict[str, float], note: str = None):
"""Init MealXML object.
Args:
name (str): name of the meal
note (str): additional information
price (dict[str, float]): prices for student, employee and other
"""
self.name = name
self.note = note
self.price = price
def xml_element(self, doc: minidom.Document):
"""Return the xml tag.
Args:
doc (minidom.Document): Working XML documnet.
"""
meal = doc.createElement("meal")
name = doc.createElement("name")
tn = doc.createTextNode(self.name)
name.appendChild(tn)
meal.appendChild(name)
if self.note is not None:
note = doc.createElement("note")
tn = doc.createTextNode(self.note)
note.appendChild(tn)
meal.appendChild(note)
for key, val in self.price.items():
if key not in ("student", "employee", "other"):
continue
price = doc.createElement("price")
price.setAttribute("role", key)
tn = doc.createTextNode(f"{val:.2f}")
price.appendChild(tn)
meal.appendChild(price)
return meal
+51
View File
@@ -0,0 +1,51 @@
from xml.dom import minidom
from dataclasses import dataclass
from stw_potsdam.xml_types.canteen_xml import CanteenXML
@dataclass
class OpenMensaXML:
"""Represents the openmensa tag in openMensaFeedv2."""
def __init__(self, version: str, canteen: CanteenXML):
"""Init OpenMensaXML.
Args:
version (str): Parser version
canteen (CanteenXML): _description_
"""
self.version = version
self.canteen = canteen
def __create_version_node(self, doc: minidom.Document):
e = doc.createElement("version")
tn = doc.createTextNode(self.version)
e.appendChild(tn)
return e
def xml_element(self, doc: minidom.Document):
"""Create openmensa XML tag.
Args:
doc (minidom.Document): Working XML document
Returns:
_type_: _description_
"""
om = doc.createElement("openmensa")
om.setAttribute("version", "2.1")
om.setAttribute("xmlns", "http://openmensa.org/open-mensa-v2")
om.setAttribute(
"xmlns:xsi", "http://www.w3.org/2001/XMLSchema-instance"
)
om.setAttribute(
"xsi:schemaLocation",
"http://openmensa.org/open-mensa-v2 "
+ "http://openmensa.org/open-mensa-v2.xsd",
)
version = self.__create_version_node(doc)
om.appendChild(version)
canteen = self.canteen.xml_element(doc)
om.appendChild(canteen)
return om
+70
View File
@@ -0,0 +1,70 @@
from dataclasses import dataclass
from xml.dom import minidom
@dataclass
class TimesXML:
"""Represents the times tag in openMensaFeedv2."""
monday: str
tuesday: str
wednesday: str
thursday: str
friday: str
saturday: str
sunday: str
def __init__(self, weekday_dict: dict[str, str] = None):
"""Init TimesXML object.
Args:
weekday_dict (dict[str, str]): _description_
"""
for key in weekday_dict:
if key in (
"monday",
"tuesday",
"wednesday",
"thursday",
"friday",
"saturday",
"sunday",
):
setattr(self, key, weekday_dict[key])
else:
raise KeyError()
def __create_node(self, doc: minidom.Document, tag: str, value: str):
e = doc.createElement(tag)
if value == "geschlossen":
e.setAttribute("closed", "true")
else:
e.setAttribute("open", value)
return e
def xml_element(self, doc: minidom.Document):
"""Return the XML representation.
Args:
doc (minidom.Document): Working XML document
Returns:
_type_: _description_
"""
times = doc.createElement("times")
times.setAttribute("type", "opening")
monday = self.__create_node(doc, "monday", self.monday)
times.appendChild(monday)
tuesday = self.__create_node(doc, "tuesday", self.tuesday)
times.appendChild(tuesday)
wednesday = self.__create_node(doc, "wednesday", self.wednesday)
times.appendChild(wednesday)
thursday = self.__create_node(doc, "thursday", self.thursday)
times.appendChild(thursday)
friday = self.__create_node(doc, "friday", self.friday)
times.appendChild(friday)
saturday = self.__create_node(doc, "saturday", self.saturday)
times.appendChild(saturday)
sunday = self.__create_node(doc, "sunday", self.sunday)
times.appendChild(sunday)
return times
+2 -2
View File
@@ -3,7 +3,7 @@ import os
import httpretty import httpretty
import pytest import pytest
from stw_potsdam.swp_webspeiseplan_api import SWP_Webspeiseplan_API from stw_potsdam.swp_webspeiseplan_api import SWPWebspeiseplanAPI
@pytest.fixture @pytest.fixture
@@ -35,7 +35,7 @@ def api_online_one_shot():
] ]
httpretty.register_uri(httpretty.POST, httpretty.register_uri(httpretty.POST,
SWP_Webspeiseplan_API.URL_BASE, SWPWebspeiseplanAPI.URL_BASE,
responses=responses) responses=responses)
httpretty.enable(allow_net_connect=False) httpretty.enable(allow_net_connect=False)
+21 -21
View File
@@ -29,44 +29,44 @@ def _read_feed(resource_name):
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_meta_consistency(): def test_meta_consistency():
raise NotImplementedError() raise NotImplementedError()
canteen = _canteen() # canteen = _canteen()
menu_feed_url = f"canteens/{canteen.key}/xml" # menu_feed_url = f"canteens/{canteen.key}/xml"
actual = feed.render_meta(canteen, menu_feed_url) # actual = feed.render_meta(canteen, menu_feed_url)
expected = _read_feed('meta_output.xml') # expected = _read_feed('meta_output.xml')
assert expected == actual # assert expected == actual
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_menu_consistency(): def test_menu_consistency():
raise NotImplementedError() raise NotImplementedError()
menu = _read_menu('input.json') # menu = _read_menu('input.json')
actual = feed.render_menu(menu) # actual = feed.render_menu(menu)
expected = _read_feed('menu_output.xml') # expected = _read_feed('menu_output.xml')
assert expected == actual # assert expected == actual
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_empty_menu(): def test_empty_menu():
raise NotImplementedError() raise NotImplementedError()
menu = _read_menu('empty.json') # menu = _read_menu('empty.json')
actual = feed.render_menu(menu) # actual = feed.render_menu(menu)
expected = _read_feed('empty_menu_output.xml') # expected = _read_feed('empty_menu_output.xml')
assert expected == actual # assert expected == actual
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_offers_dictionary(): def test_offers_dictionary():
raise NotImplementedError() raise NotImplementedError()
menu = _read_menu('offers-dict.json') # menu = _read_menu('offers-dict.json')
actual = feed.render_menu(menu) # actual = feed.render_menu(menu)
expected = _read_feed('offers-dict-output.xml') # expected = _read_feed('offers-dict-output.xml')
assert expected == actual # assert expected == actual
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_missing_category(): def test_missing_category():
raise NotImplementedError() raise NotImplementedError()
menu = _read_menu('missing-category.json') # menu = _read_menu('missing-category.json')
actual = feed.render_menu(menu) # actual = feed.render_menu(menu)
expected = _read_feed('missing-category-output.xml') # expected = _read_feed('missing-category-output.xml')
assert expected == actual # assert expected == actual
+2 -2
View File
@@ -4,9 +4,8 @@ import json
import logging import logging
import os import os
import pytest import pytest
from stw_potsdam.config import read_canteen_config from stw_potsdam.config import read_canteen_config
from stw_potsdam.views import canteen_xml_feed from stw_potsdam.views import canteen_xml_feed, app
# pragma pylint: disable=invalid-name,redefined-outer-name # pragma pylint: disable=invalid-name,redefined-outer-name
@@ -38,6 +37,7 @@ requires_online_api = pytest.mark.skipif(
@requires_online_api @requires_online_api
def test_retrieval(canteen): def test_retrieval(canteen):
try: try:
with app.app_context(), app.test_request_context():
canteen_xml_feed(canteen.key) canteen_xml_feed(canteen.key)
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
pytest.xfail('JSON endpoint returned garbage (issue #6)') pytest.xfail('JSON endpoint returned garbage (issue #6)')
+7 -10
View File
@@ -2,9 +2,6 @@
import pytest import pytest
from stw_potsdam import views from stw_potsdam import views
from flask import url_for
from tests.response_util import meal_names
# pytest fixtures are linked via parameter names of test methods # pytest fixtures are linked via parameter names of test methods
# pragma pylint: disable=unused-import,redefined-outer-name,unused-argument # pragma pylint: disable=unused-import,redefined-outer-name,unused-argument
@@ -52,14 +49,14 @@ def test_canteen_menu_api_unavailable(client, api_offline):
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_canteen_menu_request(client, api_online_one_shot): def test_canteen_menu_request(client, api_online_one_shot):
raise NotImplementedError() raise NotImplementedError()
_request_check_meals(client) # _request_check_meals(client)
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def test_canteen_menu_cached(client, api_online_one_shot): def test_canteen_menu_cached(client, api_online_one_shot):
raise NotImplementedError() raise NotImplementedError()
_request_check_meals(client) # _request_check_meals(client)
_request_check_meals(client) # _request_check_meals(client)
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
@@ -72,10 +69,10 @@ def test_canteen_menu_second_request_indeed_fails(client, api_online_one_shot):
@pytest.mark.xfail(strict=True) @pytest.mark.xfail(strict=True)
def _request_check_meals(client): def _request_check_meals(client):
raise NotImplementedError() raise NotImplementedError()
response = client.get("/canteens/griebnitzsee/xml") # response = client.get("/canteens/griebnitzsee/xml")
assert response.status_code == 200 # assert response.status_code == 200
meal = meal_names(response.data)[0] # meal = meal_names(response.data)[0]
print(meal) # print(meal)
# assert meal == "Gefüllter Germknödel \nmit Vanillesauce und Mohnzucker" # assert meal == "Gefüllter Germknödel \nmit Vanillesauce und Mohnzucker"