import os import requests import json from datetime import datetime, timedelta import sys class PMT: def __init__(self): self.session = None self.context_token = None self.user_token = None self._username = os.environ.get("USERNAME_PMT") self._password = os.environ.get("PASSWORD_PMT") self.base_url = os.environ.get("URL_PMT") def get_week_range(self, week_offset): """Get Monday date for week_offset weeks from now""" today = datetime.now() days_to_monday = 0 if today.weekday() == 0 else 7 - today.weekday() monday = today + timedelta(days=days_to_monday) target_monday = monday + timedelta(weeks=week_offset) return target_monday.strftime("%Y-%m-%d") def login(self): self.session = requests.Session() login_url = f"{self.base_url}/login" resp = self.session.get(login_url) # Step 2: SSO login POST sso_data = { "username": self._username, "password": self._password, "browserinfo": { "osname": "Linux 148", "browsername": "Firefox", "browserversion": "148", "screenresolution": "1080x1920" } } sso_resp = self.session.post(f"{self.base_url}/pmtLoginSso", json=sso_data, headers={'Content-Type': 'application/json'}) if sso_resp.status_code != 200 or not sso_resp.json().get('result', {}).get('authenticated'): print("Login failed!") print(sso_resp.text) sys.exit(1) login_data = sso_resp.json()['result'] self.context_token = login_data['context_token'] self.user_token = login_data['user_token'] print(f"āœ… Logged in as {self._username} (store AH 8541 Rhoon)") return self.session, self.context_token, self.user_token def get_shifts(self, days_ahead=14): """Fetch shifts from now to days_ahead using the REAL endpoint""" if not self.session or not self.context_token or not self.user_token: raise ValueError("Must call login() first!") shifts = [] from_date = datetime.now().strftime("%Y-%m-%d") to_date = (datetime.now() + timedelta(days=days_ahead)).strftime("%Y-%m-%d") print(f"Fetching shifts from {from_date} to {to_date}...") headers = { 'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:148.0) Gecko/20100101 Firefox/148.0', 'Accept': 'application/json, text/plain, */*', 'Accept-Language': 'en-US,en;q=0.9', 'Accept-Encoding': 'gzip, deflate, br, zstd', 'Referer': 'https://ah8541.personeelstool.nl/my-overview/my-schedule/11-2026', 'Content-Type': 'application/json', 'x-api-user': self.user_token, 'x-api-context': self.context_token, 'Connection': 'keep-alive', 'Sec-Fetch-Dest': 'empty', 'Sec-Fetch-Mode': 'cors', 'Sec-Fetch-Site': 'same-origin' } shifts_url = f"{self.base_url}/api/v3/environments/21/stores/311/employee/98104/simpleShifts" params = { 'from_date': from_date, 'to_date': to_date } resp = self.session.get(shifts_url, headers=headers, params=params) if resp.status_code == 200: data = resp.json() week_shifts = data.get('result', {}).get('shift_instances', []) shifts.extend(week_shifts) print(f"āœ… Got {len(week_shifts)} shifts!") else: print(f"āŒ Failed: {resp.status_code}") print(resp.text[:500]) return shifts @classmethod def print_shifts(cls, shifts): """Pretty print shifts (class method since it doesn't need instance)""" if not shifts: print("No upcoming shifts found.") return print("\nšŸ“… UPCOMING SHIFTS:\n") print("Date | Time | Dept | Duration | Description") print("-" * 80) for shift in sorted(shifts, key=lambda x: x.get('start_datetime', '')): start = shift.get('start_datetime', 'N/A') end = shift.get('end_datetime', 'N/A') dept = shift.get('department_name', 'N/A') duration = shift.get('duration', 'N/A') desc = shift.get('shift_remark', shift.get('description', 'N/A')) date_str = start[:10] if start != 'N/A' else 'N/A' time_str = f"{start[11:16]}-{end[11:16]}" if start != 'N/A' and end != 'N/A' else 'N/A' print(f"{date_str} | {time_str:<12} | {dept:<10} | {duration:<8} | {desc[:40]}")