diff --git a/INSTRUCTION_MANUAL.md b/INSTRUCTION_MANUAL.md new file mode 100644 index 0000000..320b25e --- /dev/null +++ b/INSTRUCTION_MANUAL.md @@ -0,0 +1,153 @@ +# Instruction Manual + +This project fetches substitution plan data from DSBMobile. + +## Important file names + +- `in.py` means `dsbapi/__init__.py` +- `run.py` is the command-line entry point + +## Setup + +Install dependencies: + +```powershell +pip install -r requirements.txt +``` + +If `pip` does not install Pillow correctly, install it manually: + +```powershell +pip install pillow +``` + +## Basic usage + +Run the script with username and password: + +```powershell +py run.py --username YOUR_USERNAME --password YOUR_PASSWORD +``` + +Example: + +```powershell +py run.py --username 161382 --password Fokus42 +``` + +You can also use environment variables: + +```powershell +$env:DSB_USERNAME='161382' +$env:DSB_PASSWORD='Fokus42' +py run.py +``` + +## Filter by type + +Show only entries where `type` is `7e`: + +```powershell +py run.py --username 161382 --password Fokus42 --type 7e +``` + +## Date logic + +`run.py` uses Germany time by default. + +Default behavior: + +- After `08:00`, it shows the next school day +- Before `08:00`, it still shows the same target school day as the previous evening +- Friday after `08:00` moves to Monday + +Examples: + +- Thursday 18:00 -> Friday plan +- Friday 07:30 -> Friday plan +- Friday 10:00 -> Monday plan + +## Timezone option + +The default timezone is hard-coded Berlin time with daylight saving support. + +You can still override it with a fixed UTC offset: + +```powershell +py run.py --username 161382 --password Fokus42 --timezone UTC+2 +py run.py --username 161382 --password Fokus42 --timezone UTC+1 +``` + +## Other options + +Set a different cutoff hour: + +```powershell +py run.py --username 161382 --password Fokus42 --cutoff-hour 9 +``` + +Force a specific date: + +```powershell +py run.py --username 161382 --password Fokus42 --date 2026-04-17 +``` + +Include image OCR: + +```powershell +py run.py --username 161382 --password Fokus42 --include-images +``` + +Use multiple options together: + +```powershell +py run.py --username 161382 --password Fokus42 --type 7e --cutoff-hour 8 +``` + +## Output + +The script prints JSON with: + +- `timezone` +- `target_date` +- `current_time` +- `filters` +- `entries` + +## What was fixed in `in.py` + +The file `dsbapi/__init__.py` was cleaned up to fix several problems: + +- missing imports +- broken image handling +- duplicate imports +- bad response parsing +- missing HTTP error handling +- broken text decoding for umlauts +- safer timetable parsing +- request timeout support + +## Common problems + +### `Unknown timezone` + +Use the default with no `--timezone`, or pass a fixed offset like `UTC+2`. + +### Broken German characters + +This was fixed in `in.py`. If text still looks wrong, run the command again and verify your terminal encoding. + +### Login or fetch failure + +Check: + +- username and password are correct +- the DSBMobile service is reachable +- your school account still has data available + +## Files + +- `dsbapi/__init__.py`: main API code +- `run.py`: CLI runner +- `requirements.txt`: dependencies +- `INSTRUCTION_MANUAL.md`: this manual diff --git a/dsbapi/__init__.py b/dsbapi/__init__.py index 4b4cdf8..89e698c 100644 --- a/dsbapi/__init__.py +++ b/dsbapi/__init__.py @@ -1,65 +1,104 @@ -# -*- coding: utf-8 -*- +# -*- coding: utf-8 -*- """ DSBApi An API for the DSBMobile substitution plan solution, which many schools use. """ -__version_info__ = ('0', '0', '14') -__version__ = '.'.join(__version_info__) -import bs4 -import json -import requests +from __future__ import annotations + +import base64 import datetime import gzip +import io +import json import uuid -import base64 + +import bs4 +import requests try: from PIL import Image -except: - import Image +except ImportError: # pragma: no cover - Pillow is the supported dependency + import Image # type: ignore[no-redef] + +try: + from pytesseract import TesseractError +except ImportError: # pragma: no cover - older pytesseract exports may differ + TesseractError = RuntimeError + import pytesseract -import requests + +__version_info__ = ("0", "0", "14") +__version__ = ".".join(__version_info__) + +DEFAULT_TABLEMAPPER = [ + "class", + "lesson", + "teacher", + "subject", + "room", + "type", + "text", +] + class DSBApi: - def __init__(self, username, password, tablemapper=['type','class','lesson','subject','room','new_subject','new_teacher','teacher']): + def __init__(self, username, password, tablemapper=None, timeout=15): """ - Class constructor for class DSBApi + Class constructor for class DSBApi. + @param username: string, the username of the DSBMobile account @param password: string, the password of the DSBMobile account - @param tablemapper: list, the field mapping of the DSBMobile tables (default: ['type','class','lesson','subject','room','new_subject','new_teacher','teacher']) - @return: class + @param tablemapper: list, the field mapping of the DSBMobile tables + @param timeout: int/float, request timeout in seconds @raise TypeError: If the attribute tablemapper is not of type list """ self.DATA_URL = "https://app.dsbcontrol.de/JsonHandler.ashx/GetData" self.username = username self.password = password + self.timeout = timeout + self.session = requests.Session() + + if tablemapper is None: + tablemapper = list(DEFAULT_TABLEMAPPER) if not isinstance(tablemapper, list): - raise TypeError('Attribute tablemapper is not of type list!') + raise TypeError("Attribute tablemapper is not of type list!") self.tablemapper = tablemapper - - # loop over tablemapper array and identify the keyword "class". The "class" will have a special operation in split up the datasets - self.class_index = None - i = 0 - while i < len(self.tablemapper): - if self.tablemapper[i] == 'class': - self.class_index = i - break - i += 1 - + self.class_index = self._find_class_index() + + def _find_class_index(self): + for index, value in enumerate(self.tablemapper): + if value == "class": + return index + return None + + def _request_json(self, url, **kwargs): + response = self.session.request(url=url, timeout=self.timeout, **kwargs) + response.raise_for_status() + return response.json() + + def _request_text(self, url): + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + if not response.encoding or response.encoding.lower() == "iso-8859-1": + response.encoding = response.apparent_encoding or "utf-8" + return response.text + + def _request_bytes(self, url): + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.content def fetch_entries(self, images=True): """ - Fetch all the DSBMobile entries - @return: list, containing lists of DSBMobile entries from the tables or only the entries if just one table was received (default: empty list) - @rais Exception: If the request to DSBMonile failed + Fetch all DSBMobile entries. + + @return: list, containing lists of DSBMobile entries from the tables or + only the entries if just one table was received + @raise Exception: If the request to DSBMobile failed """ - # Iso format is for example 2019-10-29T19:20:31.875466 - current_time = datetime.datetime.now().isoformat() - # Cut off last 3 digits and add 'Z' to get correct format - current_time = current_time[:-3] + "Z" + current_time = datetime.datetime.utcnow().isoformat(timespec="milliseconds") + "Z" - # Parameters required for the server to accept our data request params = { "UserId": self.username, "UserPw": self.password, @@ -70,121 +109,161 @@ def fetch_entries(self, images=True): "Device": "SM-G930F", "BundleId": "de.heinekingmedia.dsbmobile", "Date": current_time, - "LastUpdate": current_time + "LastUpdate": current_time, } - # Convert params into the right format - params_bytestring = json.dumps(params, separators=(',', ':')).encode("UTF-8") - params_compressed = base64.b64encode(gzip.compress(params_bytestring)).decode("UTF-8") - # Send the request + params_bytestring = json.dumps(params, separators=(",", ":")).encode("utf-8") + params_compressed = base64.b64encode(gzip.compress(params_bytestring)).decode("utf-8") json_data = {"req": {"Data": params_compressed, "DataType": 1}} - timetable_data = requests.post(self.DATA_URL, json = json_data) - - # Decompress response - data_compressed = json.loads(timetable_data.content)["d"] - data = json.loads(gzip.decompress(base64.b64decode(data_compressed))) - - # validate response before proceed - if data['Resultcode'] != 0: - raise Exception(data['ResultStatusInfo']) - - # Find the timetable page, and extract the timetable URL from it - final = [] - for page in data["ResultMenuItems"][0]["Childs"]: - for child in page["Root"]["Childs"]: - if isinstance(child["Childs"], list): - for sub_child in child["Childs"]: - final.append(sub_child["Detail"]) - else: - final.append(child["Childs"]["Detail"]) - if not final: + payload = self._request_json(self.DATA_URL, method="post", json=json_data) + + try: + data_compressed = payload["d"] + data = json.loads(gzip.decompress(base64.b64decode(data_compressed))) + except (KeyError, ValueError, OSError, TypeError) as exc: + raise Exception("Received invalid response payload from DSBMobile") from exc + + if data.get("Resultcode") != 0: + raise Exception(data.get("ResultStatusInfo", "Unknown DSBMobile error")) + + detail_urls = self._extract_detail_urls(data) + if not detail_urls: raise Exception("Timetable data could not be found") + output = [] - for entry in final: + for entry in detail_urls: if entry.endswith(".htm") and not entry.endswith(".html") and not entry.endswith("news.htm"): output.append(self.fetch_timetable(entry)) - elif entry.endswith(".jpg") and images == True: - output.append(self.fetch_img(entry)) - - final = [] - for entry in output: - if entry is not None: - final.append(entry) - - output = final + elif entry.endswith(".jpg") and images: + image_text = self.fetch_img(entry) + if image_text is not None: + output.append(image_text) if len(output) == 1: return output[0] - else: - return output + return output + + def _extract_detail_urls(self, data): + detail_urls = [] + menu_items = data.get("ResultMenuItems") or [] + if not menu_items: + return detail_urls + for page in menu_items[0].get("Childs", []): + root = page.get("Root") or {} + for child in root.get("Childs", []): + child_nodes = child.get("Childs") + if isinstance(child_nodes, list): + for sub_child in child_nodes: + detail = sub_child.get("Detail") + if detail: + detail_urls.append(detail) + elif isinstance(child_nodes, dict): + detail = child_nodes.get("Detail") + if detail: + detail_urls.append(detail) + return detail_urls def fetch_img(self, imgurl): """ - Extract data from the image + Extract OCR text from an image. + @param imgurl: string, the URL to the image - @return: list, list of dicts - @todo: Future use - implement OCR - @raise Exception: If the function will be crawled, because the funbtion is not implemented yet + @return: string or None """ - try: - img = Image.open(io.BytesIO(requests.get(imgurl))) - except: - return #haha this is quality coding surplus - - string = "" + image_bytes = self._request_bytes(imgurl) + img = Image.open(io.BytesIO(image_bytes)) + except Exception: + return None try: - return pytesseract.image_to_string(img) - except TesseractError: - raise Exception("You have to make the tesseract command accessible and work!") - return None + return pytesseract.image_to_string(img) + except TesseractError as exc: + raise Exception("You have to make the tesseract command accessible and work!") from exc def fetch_timetable(self, timetableurl): """ - parse the timetableurl HTML page and return the parsed entries + Parse the timetable HTML page and return the parsed entries. + @param timetableurl: string, the URL to the timetable in HTML format @return: list, list of dicts """ results = [] - sauce = requests.get(timetableurl).text - soupi = bs4.BeautifulSoup(sauce, "html.parser") - ind = -1 - for soup in soupi.find_all('table', {'class': 'mon_list'}): - ind += 1 - updates = [o.p.findAll('span')[-1].next_sibling.split("Stand: ")[1] for o in soupi.findAll('table', {'class': 'mon_head'})][ind] - titles = [o.text for o in soupi.findAll('div', {'class': 'mon_title'})][ind] - date = titles.split(" ")[0] - day = titles.split(" ")[1].split(", ")[0].replace(",", "") - entries = soup.find_all("tr") - entries.pop(0) - for entry in entries: - infos = entry.find_all("td") + soup = bs4.BeautifulSoup(self._request_text(timetableurl), "html.parser") + tables = soup.find_all("table", {"class": "mon_list"}) + headers = soup.find_all("table", {"class": "mon_head"}) + titles = [title.get_text(" ", strip=True) for title in soup.find_all("div", {"class": "mon_title"})] + + for index, table in enumerate(tables): + updated = self._extract_updated(headers, index) + date, day = self._extract_title_parts(titles, index) + rows = table.find_all("tr")[1:] + + for row in rows: + infos = row.find_all("td") if len(infos) < 2: continue - - # check if a "class" attribute is there, if yes, split the "class" value by "," to spread out the data rows for each school class - if self.class_index != None: - class_array = infos[self.class_index].text.split(", ") - else: - # define a dummy value if we don't have a class column (with keyword "class") - class_array = [ '---' ] - for class_ in class_array: - new_entry = dict() - new_entry["date"] = date - new_entry["day"] = day - new_entry["updated"] = updates - i = 0 - while i < len(infos): - if i < len(self.tablemapper): - attribute = self.tablemapper[i] - else: - attribute = 'col' + str(i) - if attribute == 'class': - new_entry[attribute] = class_ if infos[i].text != "\xa0" else "---" + + class_values = self._extract_class_values(infos) + for class_value in class_values: + new_entry = { + "date": date, + "day": day, + "updated": updated, + } + for col_index, info in enumerate(infos): + attribute = self.tablemapper[col_index] if col_index < len(self.tablemapper) else "col" + str(col_index) + value = info.get_text(strip=True) or "---" + if attribute == "class": + new_entry[attribute] = class_value if value != "---" else "---" else: - new_entry[attribute] = infos[i].text if infos[i].text != "\xa0" else "---" - i += 1 + new_entry[attribute] = value results.append(new_entry) return results + + def _extract_updated(self, headers, index): + if index >= len(headers): + return "---" + + spans = headers[index].find_all("span") + if not spans: + return "---" + + sibling = spans[-1].next_sibling + if not isinstance(sibling, str): + return "---" + + parts = sibling.split("Stand: ", 1) + if len(parts) == 2 and parts[1].strip(): + return parts[1].strip() + return sibling.strip() or "---" + + def _extract_title_parts(self, titles, index): + if index >= len(titles): + return "---", "---" + + title = titles[index].strip() + if not title: + return "---", "---" + + parts = title.split(" ", 1) + date = parts[0] + day_text = parts[1] if len(parts) > 1 else "---" + day = day_text.split(", ", 1)[0].replace(",", "").strip() or "---" + return date, day + + def _extract_class_values(self, infos): + if self.class_index is None or self.class_index >= len(infos): + return ["---"] + + raw_value = infos[self.class_index].get_text(strip=True) + if not raw_value: + return ["---"] + + return [part.strip() for part in raw_value.split(",") if part.strip()] or ["---"] + + +__all__ = ["DSBApi", "__version__", "__version_info__"] + + diff --git a/dsbapi/__pycache__/__init__.cpython-313.pyc b/dsbapi/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..9a19313 Binary files /dev/null and b/dsbapi/__pycache__/__init__.cpython-313.pyc differ diff --git a/requirements.txt b/requirements.txt index 037e11a..07fadde 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,4 @@ -requests -beautifulsoup4 -datetime +beautifulsoup4 +pillow pytesseract requests -PIL diff --git a/run.py b/run.py new file mode 100644 index 0000000..b6c8a17 --- /dev/null +++ b/run.py @@ -0,0 +1,214 @@ +import argparse +import json +import os +import re +import sys +from datetime import date, datetime, time, timedelta, timezone, tzinfo + +from dsbapi import DSBApi + + +DEFAULT_CUTOFF_HOUR = 8 +DEFAULT_TIMEZONE = "Europe/Berlin" +SCHOOL_WEEKDAYS = {0, 1, 2, 3, 4} +FIXED_OFFSET_PATTERN = re.compile(r"^UTC(?P[+-])(?P\d{1,2})(?::(?P\d{2}))?$", re.IGNORECASE) + + +class BerlinTimezone(tzinfo): + def tzname(self, dt): + return "CEST" if self._is_dst(dt) else "CET" + + def utcoffset(self, dt): + return timedelta(hours=2 if self._is_dst(dt) else 1) + + def dst(self, dt): + return timedelta(hours=1 if self._is_dst(dt) else 0) + + def fromutc(self, dt): + standard_time = (dt + timedelta(hours=1)).replace(tzinfo=self) + if self._is_dst(standard_time): + return (dt + timedelta(hours=2)).replace(tzinfo=self) + return standard_time + + def _is_dst(self, dt): + if dt is None: + return False + + if dt.tzinfo is not None: + dt = dt.replace(tzinfo=None) + + year = dt.year + start = self._last_sunday(year, 3, 2) + end = self._last_sunday(year, 10, 3) + return start <= dt < end + + @staticmethod + def _last_sunday(year, month, hour): + if month == 12: + next_month = date(year + 1, 1, 1) + else: + next_month = date(year, month + 1, 1) + last_day = next_month - timedelta(days=1) + while last_day.weekday() != 6: + last_day -= timedelta(days=1) + return datetime.combine(last_day, time(hour=hour)) + + +BERLIN_TZ = BerlinTimezone() + + +def build_parser(): + parser = argparse.ArgumentParser(description="Fetch DSBMobile substitution plan entries.") + parser.add_argument("--username", default=os.getenv("DSB_USERNAME"), help="DSB username") + parser.add_argument("--password", default=os.getenv("DSB_PASSWORD"), help="DSB password") + parser.add_argument( + "--tablemapper", + nargs="*", + help="Optional column mapping override, for example: --tablemapper class lesson subject", + ) + parser.add_argument( + "--type", + dest="entry_type", + help="Only include entries whose type exactly matches this value, for example: 7e", + ) + parser.add_argument( + "--timezone", + default=os.getenv("DSB_TIMEZONE", DEFAULT_TIMEZONE), + help="Timezone used for day selection. Supports Europe/Berlin and fixed offsets like UTC+2.", + ) + parser.add_argument( + "--cutoff-hour", + type=int, + default=DEFAULT_CUTOFF_HOUR, + help="Keep showing the same next school day until this hour on that day, default: 8", + ) + parser.add_argument( + "--date", + help="Optional explicit target date in YYYY-MM-DD format. Overrides automatic next-day logic.", + ) + parser.add_argument( + "--include-images", + action="store_true", + help="Run OCR on linked JPG timetable images as well.", + ) + return parser + + +def filter_entries(entries, entry_type=None, target_date=None): + if isinstance(entries, list) and entries and all(isinstance(item, dict) for item in entries): + return [entry for entry in entries if entry_matches(entry, entry_type, target_date)] + + if isinstance(entries, list): + filtered = [] + for group in entries: + if isinstance(group, list): + matches = [entry for entry in group if isinstance(entry, dict) and entry_matches(entry, entry_type, target_date)] + if matches: + filtered.append(matches) + return filtered + + return entries + + +def entry_matches(entry, entry_type=None, target_date=None): + if entry_type and entry.get("type") != entry_type: + return False + if target_date and parse_entry_date(entry.get("date")) != target_date: + return False + return True + + +def parse_entry_date(value): + if not value: + return None + + for fmt in ("%d.%m.%Y", "%d.%m.%y"): + try: + return datetime.strptime(value, fmt).date() + except ValueError: + continue + return None + + +def parse_cli_date(value): + try: + return datetime.strptime(value, "%Y-%m-%d").date() + except ValueError as exc: + raise SystemExit(f"Invalid --date value '{value}'. Use YYYY-MM-DD.") from exc + + +def resolve_timezone(name): + normalized = name.strip() + if normalized.lower() in {"europe/berlin", "berlin", "germany", "de"}: + return BERLIN_TZ, DEFAULT_TIMEZONE + + match = FIXED_OFFSET_PATTERN.match(normalized) + if match: + hours = int(match.group("hours")) + minutes = int(match.group("minutes") or 0) + if hours > 23 or minutes > 59: + raise SystemExit(f"Invalid timezone '{name}'. Use Europe/Berlin or a fixed offset like UTC+2.") + total_minutes = hours * 60 + minutes + if match.group("sign") == "-": + total_minutes *= -1 + tz = timezone(timedelta(minutes=total_minutes)) + label = f"UTC{match.group('sign')}{hours:02d}:{minutes:02d}" + return tz, label + + raise SystemExit(f"Unknown timezone '{name}'. Use Europe/Berlin or a fixed offset like UTC+2.") + + +def next_school_day(start_date): + candidate = start_date + timedelta(days=1) + while candidate.weekday() not in SCHOOL_WEEKDAYS: + candidate += timedelta(days=1) + return candidate + + +def resolve_target_date(timezone_name, cutoff_hour, explicit_date=None): + tz, timezone_label = resolve_timezone(timezone_name) + if explicit_date is not None: + return explicit_date, None, timezone_label + + now = datetime.now(timezone.utc).astimezone(tz) + anchor_date = now.date() if now.hour >= cutoff_hour else now.date() - timedelta(days=1) + return next_school_day(anchor_date), now, timezone_label + + +def main(): + parser = build_parser() + args = parser.parse_args() + + if not args.username or not args.password: + parser.error("username and password are required, either as flags or via DSB_USERNAME / DSB_PASSWORD") + if not 0 <= args.cutoff_hour <= 23: + parser.error("cutoff-hour must be between 0 and 23") + + explicit_date = parse_cli_date(args.date) if args.date else None + target_date, now, timezone_label = resolve_target_date(args.timezone, args.cutoff_hour, explicit_date) + + client = DSBApi(args.username, args.password, tablemapper=args.tablemapper) + + try: + entries = client.fetch_entries(images=args.include_images) + entries = filter_entries(entries, args.entry_type, target_date) + except Exception as exc: + print(f"Failed to fetch entries: {exc}", file=sys.stderr) + return 1 + + output = { + "timezone": timezone_label, + "target_date": target_date.isoformat(), + "current_time": now.isoformat() if now else None, + "filters": { + "type": args.entry_type, + "cutoff_hour": args.cutoff_hour, + }, + "entries": entries, + } + print(json.dumps(output, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/setup.py b/setup.py index 785a004..87a7484 100644 --- a/setup.py +++ b/setup.py @@ -1,7 +1,11 @@ +from pathlib import Path + import setuptools -with open("README.md", "r") as fh: - long_description = fh.read() + +BASE_DIR = Path(__file__).resolve().parent +long_description = (BASE_DIR / "README.md").read_text(encoding="utf-8") + setuptools.setup( name="dsbapipy", @@ -18,5 +22,11 @@ "License :: OSI Approved :: GNU General Public License v3 (GPLv3)", "Operating System :: OS Independent", ], - python_requires='>=3.5', + python_requires=">=3.8", + install_requires=[ + "beautifulsoup4", + "pillow", + "pytesseract", + "requests", + ], )