From d1c02e5b621cd4dd47ef84869b19167bb9141cc9 Mon Sep 17 00:00:00 2001 From: b267a Date: Sat, 2 Aug 2025 22:37:17 +0200 Subject: [PATCH] + Recoded entire codebase + Added proper README.md --- .github/workflows/schedule.yml | 4 +- README.md | 181 +++++++++++++++- auth/session.py | 84 ++++++++ booking/client.py | 81 +++++++ config/constants.py | 12 ++ kit.py | 374 +++++++++++++++++---------------- main.py | 87 +++----- utils/helpers.py | 14 ++ 8 files changed, 585 insertions(+), 252 deletions(-) create mode 100644 auth/session.py create mode 100644 booking/client.py create mode 100644 config/constants.py create mode 100644 utils/helpers.py diff --git a/.github/workflows/schedule.yml b/.github/workflows/schedule.yml index 718c00c..115c491 100644 --- a/.github/workflows/schedule.yml +++ b/.github/workflows/schedule.yml @@ -1,4 +1,4 @@ -name: Running Library Reservation Script +name: Daily Library Reservation Automation on: workflow_dispatch: @@ -21,4 +21,4 @@ jobs: USERNAME: ${{ secrets.USERNAME }} PASSWORD: ${{ secrets.PASSWORD }} run: | - python kit.py + python main.py diff --git a/README.md b/README.md index e04db80..0d6a758 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,179 @@ -# KIT Library Automation -Will improve the code soon, just some proof-of-concept right now! +# 📚 KIT Anny Booking Automation -I use https://cron-job.org to run the github workflow instantly without any delays. \ No newline at end of file +This Python project automates booking of study spaces or resources via the [anny.eu](https://anny.eu) platform used by the KIT (Karlsruhe Institute of Technology) library system. It logs in automatically using KIT SAML SSO, searches for available slots, and makes bookings without user interaction — ideal for recurring reservations. + +--- + +## ⚙️ Features + +- 🔐 Automatic KIT login via SAML SSO +- 📆 Configurable 3-days-ahead reservation window +- 🔎 Auto-detection of available time slots +- 🛠️ Clean and modular object-oriented codebase +- 🔁 Fully automated execution using [cron-job.org](https://cron-job.org) + GitHub API +- 📦 Easy to extend and maintain + +--- + +## 🗂️ Project Structure + +``` +anny_booking/ +├── .env # Credentials (excluded from version control) +├── main.py # Entry point for script execution +├── requirements.txt # Python dependencies +│ +├── auth/ +│ └── session.py # AnnySession class (login logic) +│ +├── booking/ +│ └── client.py # BookingClient class (resource booking) +│ +├── config/ +│ └── constants.py # API URLs and shared constants +│ +├── utils/ +│ └── helpers.py # Utility functions +│ +└── .github/ + └── workflows/ + └── schedule.yml # GitHub Actions workflow (manual trigger) +``` + +--- + +## 🚀 Setup & Installation + +### 1. Clone the repository + +```bash +git clone https://github.com/your-username/anny-booking-automation.git +cd anny-booking-automation +``` + +### 2. Set up a Python virtual environment + +```bash +python3 -m venv venv +source venv/bin/activate +pip install -r requirements.txt +``` + +### 3. Configure environment variables + +Create a `.env` file in the root of the project: + +``` +USERNAME=your_kit_username +PASSWORD=your_kit_password +``` + +> 🔒 Never commit this file to version control! + +--- + +## ⏱️ Automated Execution via cron-job.org + GitHub API + +To automate bookings (e.g. every morning), we use a hybrid approach: + +- `cron-job.org` handles flexible scheduling (e.g. every 12 minutes) +- GitHub Actions executes the actual script using a repository dispatch event + +### Why not just use `on: schedule`? + +GitHub Actions only supports fixed cron expressions (e.g. once per hour) and does **not** allow more frequent triggers like every 5 or 10 minutes. +Additionally, it suffers from two major issues: + +- ⏳ **Queue delay**: Workflows triggered via GitHub's `schedule` event are sometimes delayed by several minutes due to internal queue congestion. This can cause the booking script to miss the optimal reservation window. +- 🕒 **Timezone limitations**: GitHub's cron system uses UTC without native timezone support. That means you need to manually convert your desired local time (e.g. Europe/Berlin) and keep adjusting for daylight saving time changes. + +For these reasons, we use [cron-job.org](https://cron-job.org), which offers: + +- ✅ Precise minute-level scheduling +- ✅ Native timezone selection (e.g. Europe/Berlin) +- ✅ Immediate webhook execution with no delay + +This ensures your booking script runs at exactly the right time. + + +--- + +### How it works + +1. Set up a cron job on `cron-job.org` that sends a POST request to: + +``` +https://api.github.com/repos/your-username/anny-booking-automation/actions/workflows/schedule.yml/dispatches +``` + +2. Include the following JSON payload: + +```json +{"ref": "main"} +``` + +3. Add this header: + +``` +Authorization: Bearer YOUR_GITHUB_PERSONAL_ACCESS_TOKEN +Content-Type: application/json +Accept: application/vnd.github.v3+json +``` + +4. Your GitHub workflow (`.github/workflows/schedule.yml`) listens for this event: + +```yaml +name: Daily Library Reservation Automation + +on: + workflow_dispatch: + +jobs: + run-script: + runs-on: ubuntu-latest + steps: + - name: Checkout code + uses: actions/checkout@v2 + - name: Set up Python + uses: actions/setup-python@v2 + with: + python-version: '3.x' + - name: Install dependencies + run: | + pip install -r requirements.txt + - name: Run script + env: + USERNAME: ${{ secrets.USERNAME }} + PASSWORD: ${{ secrets.PASSWORD }} + run: | + python main.py + +``` + +> 💡 Store your KIT credentials securely as [GitHub Secrets](https://docs.github.com/en/actions/security-guides/encrypted-secrets): `USERNAME` and `PASSWORD`. + +--- + +## 📜 License + +This project is licensed under the **MIT License**. See the [LICENSE](./LICENSE) file for full details. + +--- + +## 👤 Author + +Created by @wiestju + +--- + +## 🤝 Contributing + +Pull requests are welcome! Feel free to open an issue or suggest features or improvements. + +--- + +## 📎 Related Tools + +- [cron-job.org](https://cron-job.org) +- [GitHub Actions](https://github.com/features/actions) +- [anny.eu](https://anny.eu) diff --git a/auth/session.py b/auth/session.py new file mode 100644 index 0000000..0114ba3 --- /dev/null +++ b/auth/session.py @@ -0,0 +1,84 @@ +import requests +import urllib.parse +import re +import html +from config.constants import AUTH_BASE_URL, ANNY_BASE_URL, DEFAULT_HEADERS +from utils.helpers import extract_html_value + +class AnnySession: + def __init__(self, username, password): + self.session = requests.Session() + self.username = username + self.password = password + + def login(self): + try: + self._init_headers() + self._sso_login() + self._kit_auth() + self._consume_saml() + return self.session.cookies + except Exception as e: + print(f"[Login Error] {e}") + return None + + def _init_headers(self): + self.session.headers.update({ + **DEFAULT_HEADERS, + 'accept': 'text/html, application/xhtml+xml', + 'referer': AUTH_BASE_URL + '/', + 'origin': AUTH_BASE_URL + }) + + def _sso_login(self): + r1 = self.session.get(f"{AUTH_BASE_URL}/login/sso") + self.session.headers['X-XSRF-TOKEN'] = urllib.parse.unquote(r1.cookies['XSRF-TOKEN']) + + page_data = extract_html_value(r1.text, r'data-page="(.*?)"') + version = re.search(r'"version"\s*:\s*"([a-f0-9]{32})"', page_data) + x_inertia_version = version.group(1) if version else '66b32acea13402d3aef4488ccd239c93' + + self.session.headers.update({ + 'x-requested-with': 'XMLHttpRequest', + 'x-inertia': 'true', + 'x-inertia-version': x_inertia_version + }) + + r2 = self.session.post(f"{AUTH_BASE_URL}/login/sso", json={"domain": "kit.edu"}) + redirect_url = r2.headers['x-inertia-location'] + self.redirect_response = self.session.get(redirect_url) + + def _kit_auth(self): + self.session.headers.pop('x-requested-with', None) + self.session.headers.pop('x-inertia', None) + self.session.headers.pop('x-inertia-version', None) + + csrf_token = extract_html_value(self.redirect_response.text, r'name="csrf_token" value="([^"]+)"') + + r4 = self.session.post( + 'https://idp.scc.kit.edu/idp/profile/SAML2/Redirect/SSO?execution=e1s1', + data={ + 'csrf_token': csrf_token, + 'j_username': self.username, + 'j_password': self.password, + '_eventId_proceed': '', + 'fudis_web_authn_assertion_input': '', + } + ) + + if "/consume" not in html.unescape(r4.text): + raise Exception("KIT authentication failed") + + self.saml_response_html = r4.text + + def _consume_saml(self): + consume_url = extract_html_value(self.saml_response_html, r'form action="([^"]+)"') + relay_state = extract_html_value(self.saml_response_html, r'name="RelayState" value="([^"]+)"') + saml_response = extract_html_value(self.saml_response_html, r'name="SAMLResponse" value="([^"]+)"') + + self.session.post(consume_url, data={ + 'RelayState': relay_state, + 'SAMLResponse': saml_response + }) + + self.session.get(f"{ANNY_BASE_URL}/en-us/login?target=/en-us/home?withoutIntent=true") diff --git a/booking/client.py b/booking/client.py new file mode 100644 index 0000000..cc9951d --- /dev/null +++ b/booking/client.py @@ -0,0 +1,81 @@ +import requests +from config.constants import RESOURCE_URL, BOOKING_API_BASE, CHECKOUT_FORM_API, ANNY_BASE_URL, SERVICE_ID +from utils.helpers import extract_html_value + +class BookingClient: + def __init__(self, cookies): + self.session = requests.Session() + self.session.cookies = cookies + self.token = cookies.get('anny_shop_jwt') + + self.session.headers.update({ + 'authorization': f'Bearer {self.token}', + 'content-type': 'application/vnd.api+json', + 'origin': ANNY_BASE_URL, + 'referer': ANNY_BASE_URL + '/', + 'user-agent': 'Mozilla/5.0' + }) + + def find_available_resource(self, start, end): + response = self.session.get(RESOURCE_URL, params={ + 'page[number]': 1, + 'page[size]': 250, + 'filter[available_from]': start, + 'filter[available_to]': end, + 'filter[availability_exact_match]': 1, + 'filter[availability_service_id]': SERVICE_ID, + 'filter[include_unavailable]': 0, + 'sort': 'name' + }) + resources = response.json().get('data', []) + return resources[0]['id'] if resources else None + + def reserve(self, resource_id, start, end): + booking = self.session.post( + f"{BOOKING_API_BASE}/order/bookings?include=customer&stateless=1", + json={ + "resource_id": [resource_id], + "service_id": {SERVICE_ID: 1}, + "start_date": start, + "end_date": end, + "description": "", + "customer_note": "", + "add_ons_by_service": {SERVICE_ID: [[]]}, + "sub_bookings_by_service": {}, + "strategy": "multi-resource" + } + ) + + if not booking.ok: + print("❌ Slot already taken.") + return False + + data = booking.json().get("data", {}) + oid = data.get("id") + oat = data.get("attributes", {}).get("access_token") + + checkout = self.session.get(f"{CHECKOUT_FORM_API}?oid={oid}&oat={oat}&stateless=1") + customer = checkout.json().get("default", {}).get("customer", {}) + + final = self.session.post( + f"{BOOKING_API_BASE}/order?oid={oid}&oat={oat}&stateless=1", + json={ + "customer": { + "given_name": customer.get("given_name"), + "family_name": customer.get("family_name"), + "email": customer.get("email") + }, + "accept_terms": True, + "payment_method": "", + "success_url": f"{ANNY_BASE_URL}/checkout/success?oids={oid}&oats={oat}", + "cancel_url": f"{ANNY_BASE_URL}/checkout?step=checkout&childResource={resource_id}", + "meta": {"timezone": "Europe/Berlin"} + } + ) + + if final.ok: + print("✅ Reservation successful!") + return True + + print("❌ Reservation failed.") + return False diff --git a/config/constants.py b/config/constants.py new file mode 100644 index 0000000..9ed9aae --- /dev/null +++ b/config/constants.py @@ -0,0 +1,12 @@ +AUTH_BASE_URL = "https://auth.anny.eu" +ANNY_BASE_URL = "https://anny.eu" +BOOKING_API_BASE = "https://b.anny.eu/api/v1" +CHECKOUT_FORM_API = "https://b.anny.eu/api/ui/checkout-form" +RESOURCE_URL = f"{BOOKING_API_BASE}/resources/1-lehrbuchsammlung-eg-und-1-og/children" +SERVICE_ID = "449" + +DEFAULT_HEADERS = { + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0', + 'accept': 'application/vnd.api+json', + 'accept-encoding': 'plain' +} \ No newline at end of file diff --git a/kit.py b/kit.py index dfdf40a..c7e3d13 100644 --- a/kit.py +++ b/kit.py @@ -1,226 +1,232 @@ -import requests -import urllib.parse +import os import re import html -import time -import datetime import pytz -import os +import requests +import datetime +import urllib.parse from dotenv import load_dotenv -def get_day(): - dt = datetime.datetime.now(pytz.timezone('Europe/Berlin')) + datetime.timedelta(days=3) - date = dt.strftime('%Y-%m-%d') - return date -def login(username, password): +# === CONFIGURATION CONSTANTS === # +AUTH_BASE_URL = "https://auth.anny.eu" +ANNY_BASE_URL = "https://anny.eu" +BOOKING_API_BASE = "https://b.anny.eu/api/v1" +CHECKOUT_FORM_API = "https://b.anny.eu/api/ui/checkout-form" +RESOURCE_URL = f"{BOOKING_API_BASE}/resources/1-lehrbuchsammlung-eg-und-1-og/children" +SERVICE_ID = "449" - session = requests.Session() +DEFAULT_HEADERS = { + 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0', + 'accept': 'application/vnd.api+json', + 'accept-encoding': 'plain' +} - session.headers = { - 'accept': 'text/html, application/xhtml+xml', - 'accept-encoding': 'plain', - 'referer': 'https://auth.anny.eu/', - 'origin': 'https://auth.anny.eu', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0', - } - - r = session.get( - 'https://auth.anny.eu/login/sso' - ) - - session.headers['X-XSRF-TOKEN'] = urllib.parse.unquote(r.cookies['XSRF-TOKEN']) - - match = re.search(r'data-page="(.*?)"', r.text) - if match: - decoded = html.unescape(match.group(1)) - version_match = re.search(r'"version"\s*:\s*"([a-f0-9]{32})"', decoded) - if version_match: - x_inertia_version = version_match.group(1) - else: - x_inertia_version = '66b32acea13402d3aef4488ccd239c93' +def get_future_datetime(days_ahead=3, hour="13:00:00"): + dt = datetime.datetime.now(pytz.timezone('Europe/Berlin')) + datetime.timedelta(days=days_ahead) + return dt.strftime(f"%Y-%m-%dT{hour}+02:00") +def extract_html_value(text, pattern): + match = re.search(pattern, text) + if not match: + raise ValueError(f"Pattern not found: {pattern}") + return html.unescape(match.group(1)) - session.headers.update( - { + +class AnnySession: + def __init__(self, username, password): + self.session = requests.Session() + self.username = username + self.password = password + self.cookies = None + + def login(self): + try: + self._init_headers() + self._sso_login() + self._kit_auth() + self._consume_saml() + self.cookies = self.session.cookies + print("✅ Login successful") + return self.cookies + except Exception as e: + print(f"❌ Login failed: {e}") + return None + + def _init_headers(self): + self.session.headers.update({ + **DEFAULT_HEADERS, + 'accept': 'text/html, application/xhtml+xml', + 'referer': AUTH_BASE_URL + '/', + 'origin': AUTH_BASE_URL + }) + + def _sso_login(self): + r1 = self.session.get(f"{AUTH_BASE_URL}/login/sso") + self.session.headers['X-XSRF-TOKEN'] = urllib.parse.unquote(r1.cookies['XSRF-TOKEN']) + + page_data = extract_html_value(r1.text, r'data-page="(.*?)"') + version = re.search(r'"version"\s*:\s*"([a-f0-9]{32})"', page_data) + x_inertia_version = version.group(1) if version else '66b32acea13402d3aef4488ccd239c93' + + self.session.headers.update({ 'x-requested-with': 'XMLHttpRequest', 'x-inertia': 'true', - 'x-inertia-version': x_inertia_version, - } - ) + 'x-inertia-version': x_inertia_version + }) - r2 = session.post( - 'https://auth.anny.eu/login/sso', - json={ - 'domain': 'kit.edu' - }, headers={ - 'referer': 'https://auth.anny.eu/login/sso', - } - ) + r2 = self.session.post(f"{AUTH_BASE_URL}/login/sso", json={"domain": "kit.edu"}) + redirect_url = r2.headers['x-inertia-location'] + self.redirect_response = self.session.get(redirect_url) - redirect_url = r2.headers['x-inertia-location'] + def _kit_auth(self): + self.session.headers.pop('x-requested-with', None) + self.session.headers.pop('x-inertia', None) + self.session.headers.pop('x-inertia-version', None) - r3 = session.get( - redirect_url, - allow_redirects=True - ) + csrf_token = extract_html_value(self.redirect_response.text, r'name="csrf_token" value="([^"]+)"') - session.headers.pop('x-requested-with') - session.headers.pop('x-inertia') - session.headers.pop('x-inertia-version') + r4 = self.session.post( + 'https://idp.scc.kit.edu/idp/profile/SAML2/Redirect/SSO?execution=e1s1', + data={ + 'csrf_token': csrf_token, + 'j_username': self.username, + 'j_password': self.password, + '_eventId_proceed': '', + 'fudis_web_authn_assertion_input': '', + } + ) - pattern = r'name="csrf_token" value="([^"]+)"' - csrf_token = re.search(pattern, r3.text).group(1) + if "/consume" not in html.unescape(r4.text): + raise Exception("KIT authentication failed") - r4 = session.post( - 'https://idp.scc.kit.edu/idp/profile/SAML2/Redirect/SSO?execution=e1s1', - data={ - 'csrf_token': csrf_token, - 'j_username': username, - 'j_password': password, - '_eventId_proceed': '', - 'fudis_web_authn_assertion_input': '', - } - ) + self.saml_response_html = r4.text - response = html.unescape(r4.text) + def _consume_saml(self): + consume_url = extract_html_value(self.saml_response_html, r'form action="([^"]+)"') + relay_state = extract_html_value(self.saml_response_html, r'name="RelayState" value="([^"]+)"') + saml_response = extract_html_value(self.saml_response_html, r'name="SAMLResponse" value="([^"]+)"') - if '/consume' in response: - print("KIT-Login successful!") - else: - print("Failed to login, probably wrong credentials!") - return False + self.session.post(consume_url, data={ + 'RelayState': relay_state, + 'SAMLResponse': saml_response + }) - pattern = r'form action="([^"]+)"' - consume_url = re.search(pattern, response).group(1) - pattern = r'name="RelayState" value="([^"]+)"' - relayState = re.search(pattern, response).group(1) - pattern = r'name="SAMLResponse" value="([^"]+)"' - samlResponse = re.search(pattern, response).group(1) - - r5 = session.post( - consume_url, - data={ - 'RelayState': relayState, - 'SAMLResponse': samlResponse - } - ) - - r6 = session.get( - 'https://anny.eu/en-us/login?target=/en-us/home?withoutIntent=true', - allow_redirects=True - ) + self.session.get(f"{ANNY_BASE_URL}/en-us/login?target=/en-us/home?withoutIntent=true") - return session.cookies - -def test_reservation(): - load_dotenv('credentials.env', override=True) - - username = os.getenv('USERNAME') - password = os.getenv('PASSWORD') - - cookies = login(username, password) - - TOKEN = cookies['anny_shop_jwt'] - - ses = requests.Session() - ses.cookies = cookies - ses.headers = { - 'accept': 'application/vnd.api+json', - 'accept-encoding': 'plain', - 'authorization': 'Bearer ' + TOKEN, +class BookingClient: + def __init__(self, cookies): + self.session = requests.Session() + self.session.cookies = cookies + self.token = cookies.get('anny_shop_jwt') + self.session.headers.update({ + **DEFAULT_HEADERS, + 'authorization': f'Bearer {self.token}', 'content-type': 'application/vnd.api+json', - 'origin': 'https://anny.eu', - 'referer': 'https://anny.eu/', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0' - } + 'origin': ANNY_BASE_URL, + 'referer': ANNY_BASE_URL + '/' + }) - start_date = get_day() + "T13:00:00+02:00" - end_date = get_day() + "T18:00:00+02:00" - - pre = ses.get( - 'https://b.anny.eu/api/v1/resources/1-lehrbuchsammlung-eg-und-1-og/children', - params={ + def find_available_resource(self, start, end): + response = self.session.get(RESOURCE_URL, params={ 'page[number]': 1, 'page[size]': 250, - 'filter[available_from]': start_date, - 'filter[available_to]': end_date, + 'filter[available_from]': start, + 'filter[available_to]': end, 'filter[availability_exact_match]': 1, 'filter[exclude_hidden]': 0, 'filter[exclude_child_resources]': 0, - 'filter[availability_service_id]': 449, + 'filter[availability_service_id]': SERVICE_ID, 'filter[include_unavailable]': 0, 'filter[pre_order_ids]': '', 'sort': 'name' - } - ) + }) - ressources = pre.json()['data'] + resources = response.json().get('data', []) + return resources[0]['id'] if resources else None - if len(ressources) <= 0: - print("No slots available anymore!") - return False - - ressource_id = ressources[0]['id'] - - r = ses.post( - 'https://b.anny.eu/api/v1/order/bookings?include=customer,voucher,bookings.booking_add_ons.add_on.cover_image,bookings.sub_bookings.resource,bookings.sub_bookings.service,bookings.series_bookings,bookings.customer,bookings.service.custom_forms.custom_fields,bookings.cancellation_policy,bookings.resource.cover_image,bookings.resource.parent,bookings.resource.category,bookings.reminders,bookings.booking_series,sub_orders.bookings,sub_orders.organization.legal_documents&stateless=1', - json={ - "resource_id": [ - ressource_id - ], "service_id": { - "449": 1 - }, "start_date": start_date, - "end_date": end_date, - "description":"", "customer_note": "", - "add_ons_by_service": { - "449": [ - [] - ] - }, "sub_bookings_by_service": {}, - "strategy": "multi-resource" - }, cookies=cookies - ) - - if not r.ok: - print("Slot is not available anymore!") - return False - - oid = r.json()['data']['id'] - oat = r.json()['data']['attributes']['access_token'] - - r2 = ses.get( - 'https://b.anny.eu/api/ui/checkout-form?oid=' + oid + '&oat=' + oat + '&stateless=1' - ) - - customer = r2.json()['default']['customer'] - - r3 = ses.post( - 'https://b.anny.eu/api/v1/order?include=customer,voucher,bookings.booking_add_ons.add_on,bookings.sub_bookings.resource,bookings.sub_bookings.service,bookings.service.custom_forms.custom_fields,bookings.cancellation_policy,bookings.resource.cover_image,bookings.resource.parent,bookings.reminders,bookings.customer,bookings.attendees,sub_orders.bookings,sub_orders.organization.legal_documents,last_payment.method&oid=' + oid + '&oat=' + oat + '&stateless=1', - json={ - "customer": { - "given_name": customer['given_name'], - "family_name": customer['family_name'], - "email": customer['email'] - }, "accept_terms": True, - "payment_method": "", - "success_url": "https://anny.eu/checkout/success?oids=" + oid + "&oats=" + oat, - "cancel_url": "https://anny.eu/checkout?step=checkout&childResource=3302", - "meta": { - "timezone":"Europe/Berlin" + def reserve(self, resource_id, start, end): + booking = self.session.post( + f"{BOOKING_API_BASE}/order/bookings?include=customer&stateless=1", + json={ + "resource_id": [resource_id], + "service_id": {SERVICE_ID: 1}, + "start_date": start, + "end_date": end, + "description": "", + "customer_note": "", + "add_ons_by_service": {SERVICE_ID: [[]]}, + "sub_bookings_by_service": {}, + "strategy": "multi-resource" } - } - ) + ) - if r3.ok: - print("Reservation successful!") - return True + if not booking.ok: + print("❌ Slot already taken.") + return False - print("Reservation failed!") - return False + data = booking.json().get("data", {}) + oid = data.get("id") + oat = data.get("attributes", {}).get("access_token") -test_reservation() + checkout = self.session.get(f"{CHECKOUT_FORM_API}?oid={oid}&oat={oat}&stateless=1") + customer = checkout.json().get("default", {}).get("customer", {}) + + final = self.session.post( + f"{BOOKING_API_BASE}/order?oid={oid}&oat={oat}&stateless=1", + json={ + "customer": { + "given_name": customer.get("given_name"), + "family_name": customer.get("family_name"), + "email": customer.get("email") + }, + "accept_terms": True, + "payment_method": "", + "success_url": f"{ANNY_BASE_URL}/checkout/success?oids={oid}&oats={oat}", + "cancel_url": f"{ANNY_BASE_URL}/checkout?step=checkout&childResource={resource_id}", + "meta": {"timezone": "Europe/Berlin"} + } + ) + + if final.ok: + print("✅ Reservation successful!") + return True + + print("❌ Reservation failed.") + return False + + +def main(): + load_dotenv('credentials.env', override=True) + username = os.getenv("USERNAME") + password = os.getenv("PASSWORD") + + if not username or not password: + print("❌ Missing credentials in .env") + return + + session = AnnySession(username, password) + cookies = session.login() + + if not cookies: + return + + booking = BookingClient(cookies) + start = get_future_datetime(hour="13:00:00") + end = get_future_datetime(hour="18:00:00") + + resource_id = booking.find_available_resource(start, end) + + if not resource_id: + print("⚠️ No available resources found.") + return + + booking.reserve(resource_id, start, end) + + +if __name__ == "__main__": + main() diff --git a/main.py b/main.py index 23fb6ba..8302762 100644 --- a/main.py +++ b/main.py @@ -1,73 +1,34 @@ -import requests -import urllib.parse import os from dotenv import load_dotenv +from auth.session import AnnySession +from booking.client import BookingClient +from utils.helpers import get_future_datetime -import assets.constants +def main(): + load_dotenv('.env', override=True) + username = os.getenv("USERNAME") + password = os.getenv("PASSWORD") -class Bibliothek: - def __init__(self): - self.session = requests.Session() + if not username or not password: + print("❌ Missing USERNAME or PASSWORD in .env") + return - self.session.headers = { - 'accept': 'application/vnd.api+json', - 'accept-encoding': 'gzip, deflate, br, zstd', - 'accept-language': 'de', - 'content-type': 'text/html; charset=utf-8', - 'origin': 'https://anny.eu', - 'referer': 'https://anny.eu/', - 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:140.0) Gecko/20100101 Firefox/140.0' - } + session = AnnySession(username, password) + cookies = session.login() - def login(self): - load_dotenv() - username = os.getenv('USERNAME') - password = os.getenv('PASSWORD') - login = KIT(username, password) + if not cookies: + return - self.session.headers['authorization'] = ("Bearer " + TOKEN).encode('utf-8') + booking = BookingClient(cookies) + start = get_future_datetime(hour="13:00:00") + end = get_future_datetime(hour="18:00:00") - def get_intervals(self, ressource_id, date): + resource_id = booking.find_available_resource(start, end) - r = self.session.get( - 'https://b.anny.eu/api/v1/intervals/start', - params={ - 'date': date, - 'service_id[449]': 1, - 'ressource_id': ressource_id, - 'timezone': assets.constants.TIMEZONE - } - ) - return r.json() + if resource_id: + booking.reserve(resource_id, start, end) + else: + print("⚠️ No available slots found.") - def calculate_all_available_slots(self, ressource_id, date): - - slot_list = self.get_intervals(ressource_id, date) - - for slot in slot_list: - message = slot['start_date'] + " - " + str(slot['number_available']) - print(message) - - def get_children(self, ressource = '1-lehrbuchsammlung-eg-und-1-og'): - - r = self.session.get( - 'https://b.anny.eu/api/v1/resources/' + ressource + '/children', - params={ - 'page[number]': 1, - 'page[size]': 1000, - 'filter[available_from]': '2025-07-10T00:00:00+02:00', - # 'filter[availability_exact_match]': 0, - 'filter[exclude_hidden]': 0, - 'filter[exclude_child_resources]': 0, - 'filter[availability_service_id]': 449, - 'filter[include_unavailable]': 1, - 'sort': 'name', - } - ) - - return r.json() - - -if __name__ == '__main__': - bibliothek = Bibliothek() - print(bibliothek.get_children()) +if __name__ == "__main__": + main() diff --git a/utils/helpers.py b/utils/helpers.py new file mode 100644 index 0000000..1e4899f --- /dev/null +++ b/utils/helpers.py @@ -0,0 +1,14 @@ +import re +import html +import datetime +import pytz + +def get_future_datetime(days_ahead=3, hour="13:00:00"): + dt = datetime.datetime.now(pytz.timezone('Europe/Berlin')) + datetime.timedelta(days=days_ahead) + return dt.strftime(f"%Y-%m-%dT{hour}+02:00") + +def extract_html_value(text, pattern): + match = re.search(pattern, text) + if not match: + raise ValueError(f"Pattern not found: {pattern}") + return html.unescape(match.group(1))