commit a16b90e1816f550af13c9276dd60bf5cfc1ca104 Author: MarcUs7i <96580944+MarcUs7i@users.noreply.github.com> Date: Sat May 24 01:03:16 2025 +0200 Initial Commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..df1ac3a --- /dev/null +++ b/.gitignore @@ -0,0 +1,184 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# UV +# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +#uv.lock + +# poetry +# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control. +# This is especially recommended for binary packages to ensure reproducibility, and is more +# commonly ignored for libraries. +# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control +#poetry.lock + +# pdm +# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control. +#pdm.lock +# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it +# in version control. +# https://pdm.fming.dev/latest/usage/project/#working-with-version-control +.pdm.toml +.pdm-python +.pdm-build/ + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ + +# PyCharm +# JetBrains specific template is maintained in a separate JetBrains.gitignore that can +# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore +# and can be added to the global gitignore or merged into this file. For a more nuclear +# option (not recommended) you can uncomment the following to ignore the entire idea folder. +#.idea/ + +# Visual Studio Code +# Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore +# that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore +# and can be added to the global gitignore or merged into this file. However, if you prefer, +# you could uncomment the following to ignore the enitre vscode folder +# .vscode/ + +# Ruff stuff: +.ruff_cache/ + +# PyPI configuration file +.pypirc + +# Library testing stuff +test-lib.py \ No newline at end of file diff --git a/LidlConnect/__init__.py b/LidlConnect/__init__.py new file mode 100644 index 0000000..16f005e --- /dev/null +++ b/LidlConnect/__init__.py @@ -0,0 +1,5 @@ +"""Lidl Connect API client library.""" + +from .client import LidlConnect + +__all__ = ["LidlConnect"] \ No newline at end of file diff --git a/LidlConnect/api/__init__.py b/LidlConnect/api/__init__.py new file mode 100644 index 0000000..7f59b81 --- /dev/null +++ b/LidlConnect/api/__init__.py @@ -0,0 +1 @@ +"""API modules for Lidl Connect.""" \ No newline at end of file diff --git a/LidlConnect/api/invoices.py b/LidlConnect/api/invoices.py new file mode 100644 index 0000000..d1bf123 --- /dev/null +++ b/LidlConnect/api/invoices.py @@ -0,0 +1,173 @@ +"""Invoices-related API functionality for Lidl Connect.""" + +from typing import Dict, Any, List, Optional +from datetime import datetime +from ..helpers import ttl_cache + +class InvoicesMixin: + """Invoices-related API methods for Lidl Connect API.""" + + INVOICES_URL = "https://selfcare.lidl-connect.at/customer/invoices/invoice-list" + VOUCHER_URL = "https://selfcare.lidl-connect.at/customer/invoices/consumed-vouchers" + + @ttl_cache(30) + def get_invoices(self) -> List[Dict[str, Any]]: + """ + Get list of invoices for the current account. + + Returns: + List[Dict]: List of invoice objects with transaction details + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + "ocrAvailable": "0", + } + + payload = {"accountId": self.endpoint_id, "userId": self.user_id, "endpointId": self.endpoint_id} + r = self.session.post(self.INVOICES_URL, headers=headers, json=payload) + + if r.status_code != 200: + if r.status_code == 422 and r.text == '[]': + return [] + raise ValueError(f"Invoices request failed: {r.status_code} {r.text!r}") + + return r.json() + + @ttl_cache(30) + def get_vouchers(self) -> List[Dict[str, Any]]: + """ + Get list of consumed vouchers for the current account. + + Returns: + List[Dict]: List of voucher objects with transaction details + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + "ocrAvailable": "0", + } + + payload = {"userId": self.user_id, "endpointId": self.endpoint_id} + r = self.session.post(self.VOUCHER_URL, headers=headers, json=payload) + + if r.status_code != 200: + if r.status_code == 422 and r.text == '[]': + return [] + raise ValueError(f"Vouchers request failed: {r.status_code} {r.text!r}") + + return r.json() + + def print_invoices(self) -> None: + """ + Pretty-print invoice and voucher history. + """ + try: + invoices = self.get_invoices() + vouchers = self.get_vouchers() + + if not invoices and not vouchers: + print("\nNo payment history available") + return + + if invoices: + print(f"\n{'=' * 20} INVOICE HISTORY {'=' * 20}") + + for invoice in invoices: + try: + posting_date = datetime.fromisoformat(invoice.get("postingDate").replace("Z", "+00:00")) + date_str = posting_date.strftime("%d %b %Y, %H:%M") + except (ValueError, AttributeError): + date_str = invoice.get("postingDate", "Unknown date") + + payment_type = invoice.get("type", "").capitalize() + provider = invoice.get("provider", "") + channel = invoice.get("channel", "").replace("_", " ").title() + payment_method = f"{payment_type} via {provider} ({channel})" + + print(f"\n{'-' * 60}") + print(f"Transaction ID: {invoice.get('id')}") + print(f"Date: {date_str}") + print(f"Amount: €{invoice.get('amount', 0):.2f}") + print(f"Payment Method: {payment_method}") + + if vouchers: + print(f"\n{'=' * 20} VOUCHER HISTORY {'=' * 20}") + + for voucher in vouchers: + try: + posting_date = datetime.fromisoformat(voucher.get("consumedDate").replace("Z", "+00:00")) + date_str = posting_date.strftime("%d %b %Y, %H:%M") + except (ValueError, AttributeError): + date_str = voucher.get("consumedDate", "Unknown date") + + print(f"\n{'-' * 60}") + print(f"Voucher ID: {voucher.get('id')}") + print(f"Serial: {voucher.get('serial')}") + print(f"Value: €{voucher.get('balanceAdvice', 0):.2f}") + print(f"Consumed Date: {date_str}") + except Exception as e: + print(f"\nError fetching payment history: {str(e)}") + import traceback + traceback.print_exc() + + def get_total_spent(self) -> float: + """ + Calculate the total amount spent across all invoices and vouchers. + + Returns: + float: Total amount in euros + """ + total = 0.0 + try: + invoices = self.get_invoices() + vouchers = self.get_vouchers() + total += sum(invoice.get("amount", 0) for invoice in invoices) + total += sum(voucher.get("balanceAdvice", 0) for voucher in vouchers) + except Exception: + pass + return total + + @property + def last_payment_date(self) -> Optional[str]: + """ + Get the date of the most recent payment from either invoices or vouchers. + + Returns: + str: ISO formatted date string or None if no payments + """ + try: + all_dates = [] + + invoices = self.get_invoices() + invoice_dates = [(invoice.get("postingDate"), invoice.get("amount")) + for invoice in invoices if invoice.get("postingDate")] + + vouchers = self.get_vouchers() + voucher_dates = [(voucher.get("consumedDate"), voucher.get("balanceAdvice")) + for voucher in vouchers if voucher.get("consumedDate")] + + all_dates = invoice_dates + voucher_dates + if not all_dates: + return None + + all_dates.sort(key=lambda x: x[0], reverse=True) + return all_dates[0][0] + except Exception: + return None \ No newline at end of file diff --git a/LidlConnect/api/tariffs.py b/LidlConnect/api/tariffs.py new file mode 100644 index 0000000..ddf4b0f --- /dev/null +++ b/LidlConnect/api/tariffs.py @@ -0,0 +1,138 @@ +"""Tariffs-related API functionality for Lidl Connect.""" + +from typing import Dict, Any, List +from ..helpers import ttl_cache + +class TariffsMixin: + """Tariffs-related API methods for Lidl Connect API.""" + + TARIFFS_URL = "https://selfcare.lidl-connect.at/customer/tariffs/all-packages" + + @ttl_cache(60) + def get_tariffs(self) -> List[Dict[str, Any]]: + """ + Get all available tariffs for the current account. + + Returns: + List[Dict]: List of tariff objects with relevant information + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + } + + payload = {"userId": self.user_id, "endpointId": self.endpoint_id} + r = self.session.post(self.TARIFFS_URL, headers=headers, json=payload) + + if r.status_code != 200: + raise ValueError(f"Tariffs request failed: {r.status_code} {r.text!r}") + + data = r.json() + tariffs = [item for item in data if item.get("category") == "TARIFF"] + + result = [] + for tariff in tariffs: + processed_tariff = { + "id": tariff.get("id"), + "itemId": tariff.get("itemId"), + "name": tariff.get("name"), + "description": tariff.get("description"), + "sort": tariff.get("sort", 0), + "featured": tariff.get("featured", False), + "visible": tariff.get("visible", True), + } + + translations = tariff.get("translations") + if translations and isinstance(translations, dict): + german_translation = translations.get("de", {}) + if german_translation and isinstance(german_translation, dict): + processed_tariff["german_name"] = german_translation.get("name") + processed_tariff["german_content"] = german_translation.get("content") + processed_tariff["german_teaser"] = german_translation.get("teaser") + + result.append(processed_tariff) + + result.sort(key=lambda x: x.get("sort", 0)) + return result + + def print_tariffs(self) -> None: + """ + Pretty-print available tariffs. + """ + try: + tariffs = self.get_tariffs() + + if not tariffs: + print("\nNo tariff information available") + return + + print(f"\n{'=' * 20} AVAILABLE TARIFFS {'=' * 20}") + for tariff in tariffs: + print(f"\n{'-' * 60}") + print(f"Name: {tariff.get('german_name') or tariff.get('name')}") + print(f"ID: {tariff.get('itemId')}") + + if tariff.get('german_teaser'): + import html + import re + + # HTML entities + teaser = html.unescape(tariff.get('german_teaser')) + + # style attributes + teaser = re.sub(r'style="[^"]*"', '', teaser) + teaser = re.sub(r'class="[^"]*"', '', teaser) + + # table structure + teaser = re.sub(r']*>', '\n', teaser) + teaser = re.sub(r']*>', ' | ', teaser) + teaser = re.sub(r'', '', teaser) + teaser = re.sub(r'', '', teaser) + + # line breaks + teaser = teaser.replace('
', '\n').replace('
', '\n') + teaser = teaser.replace('

', '').replace('

', '\n') + + # divs and spans + teaser = re.sub(r']*>', '', teaser) + teaser = re.sub(r'', '', teaser) + teaser = re.sub(r']*>', '', teaser) + teaser = re.sub(r'', '', teaser) + + # other HTML tags + teaser = re.sub(r'<[^>]+>', '', teaser) + + # whitespace and cleanup + teaser = re.sub(r'\s+', ' ', teaser) + teaser = re.sub(r' \| ', ' | ', teaser) + teaser = re.sub(r'\n\s+', '\n', teaser) + + # newlines + teaser = re.sub(r'\n+', '\n', teaser) + + # special chars + teaser = teaser.replace('\\u00a0', ' ') + + # keep only printable ASCII chars + teaser = ''.join([i if ord(i) < 128 else ' ' for i in teaser]) + + # other table tags + for tag in ['', '
', '', '']: + teaser = teaser.replace(tag, '') + + print(f"\nDetails:\n{teaser.strip()}") + + print(f"Featured: {'✓' if tariff.get('featured') else '✗'}") + print(f"Visible: {'✓' if tariff.get('visible') else '✗'}") + except Exception as e: + print(f"\nError fetching tariffs: {str(e)}") + import traceback + traceback.print_exc() \ No newline at end of file diff --git a/LidlConnect/api/usage.py b/LidlConnect/api/usage.py new file mode 100644 index 0000000..4c88cd5 --- /dev/null +++ b/LidlConnect/api/usage.py @@ -0,0 +1,197 @@ +"""Usage-related API functionality for Lidl Connect.""" + +from typing import Dict, Any, Optional +from ..helpers import ttl_cache + +class UsageMixin: + """Usage-related API methods for Lidl Connect.""" + + USAGE_URL = "https://selfcare.lidl-connect.at/customer/usage/" + + @ttl_cache(5) + def get_usage_data(self) -> Dict[str, Any]: + """ + Get usage data for the current account. + + Returns: + Dict: Usage data including instanceGroups with counters + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + } + payload = {"userId": self.user_id, "endpointId": self.endpoint_id} + + r = self.session.post(self.USAGE_URL, headers=headers, json=payload) + if r.status_code != 200: + raise ValueError(f"Usage request failed: {r.status_code} {r.text!r}") + + return r.json() + + def print_usage_summary(self, data: Optional[Dict[str, Any]] = None) -> None: + """ + Pretty-print usage summary data. + + Args: + data: Optional usage data. If None, will fetch new data + """ + if data is None: + data = self.get_usage_data() + + for group in data.get("instanceGroups", []): + print(f"{group['itemName']} ({group['itemCategory']})") + for elem in group.get("instanceElements", []): + print(f" • Valid: {elem['validFrom']} → {elem['validTo']}") + for counter in elem.get("counters", []): + nv = counter["niceValue"] + unit = nv.get("type") or counter.get("baseValue", {}).get("type", "") + print(f" - {counter['counterId']}: {nv['value']} / {nv['initialValue']} {unit}") + print() + + def get_remaining_data(self) -> Dict[str, float]: + """ + Get remaining data balance (in GiB). + + Returns: + Dict with remaining, total, and used data in GiB + """ + data = self.get_usage_data() + result = {"remaining": 0, "total": 0, "used": 0} + + for group in data.get("instanceGroups", []): + for elem in group.get("instanceElements", []): + for counter in elem.get("counters", []): + if counter["counterId"] == "DATA": + nv = counter["niceValue"] + if nv.get("type") == "GiB": + result["remaining"] = nv["value"] + result["total"] = nv["initialValue"] + result["used"] = nv["initialValue"] - nv["value"] + + return result + + def get_remaining_eu_data(self) -> Dict[str, float]: + """ + Get remaining EU data balance (in GiB). + + Returns: + Dict with remaining, total, and used EU data in GiB + """ + data = self.get_usage_data() + result = {"remaining": 0, "total": 0, "used": 0} + + for group in data.get("instanceGroups", []): + for elem in group.get("instanceElements", []): + for counter in elem.get("counters", []): + if counter["counterId"] == "DATA_EU": + nv = counter["niceValue"] + if nv.get("type") == "GiB": + result["remaining"] = nv["value"] + result["total"] = nv["initialValue"] + result["used"] = nv["initialValue"] - nv["value"] + + return result + + def get_remaining_minutes(self) -> Dict[str, float]: + """ + Get remaining voice minutes. + + Returns: + Dict with remaining, total, and used minutes + """ + data = self.get_usage_data() + result = {"remaining": 0, "total": 0, "used": 0} + + for group in data.get("instanceGroups", []): + for elem in group.get("instanceElements", []): + for counter in elem.get("counters", []): + if counter["counterId"] == "VOICE_SMS": + nv = counter["niceValue"] + if nv.get("type") == "MIN": + result["remaining"] = nv["value"] + result["total"] = nv["initialValue"] + result["used"] = nv["initialValue"] - nv["value"] + + return result + + @property + def tariff_package_valid_from(self) -> Optional[str]: + """ + Get the start date of the current tariff package. + + Returns: + ISO formatted date string or None if not available + """ + try: + data = self.get_usage_data() + for group in data.get("instanceGroups", []): + if group.get("itemCategory") == "TARIFF_PACKAGE": + for elem in group.get("instanceElements", []): + return elem.get("validFrom") + return None + except Exception: + return None + + @property + def tariff_package_valid_to(self) -> Optional[str]: + """ + Get the end date of the current tariff package. + + Returns: + ISO formatted date string or None if not available + """ + try: + data = self.get_usage_data() + for group in data.get("instanceGroups", []): + if group.get("itemCategory") == "TARIFF_PACKAGE": + for elem in group.get("instanceElements", []): + return elem.get("validTo") + return None + except Exception: + return None + + @property + def tariff_package_details(self) -> Optional[Dict[str, Any]]: + """ + Get detailed information about the current tariff package. + + Returns: + Dict containing name, category, validFrom, validTo and other details + or None if not available + """ + try: + data = self.get_usage_data() + for group in data.get("instanceGroups", []): + if group.get("itemCategory") == "TARIFF_PACKAGE": + result = { + "name": group.get("itemName"), + "category": group.get("itemCategory"), + } + + if group.get("instanceElements") and len(group.get("instanceElements")) > 0: + elem = group.get("instanceElements")[0] + result.update({ + "validFrom": elem.get("validFrom"), + "validTo": elem.get("validTo"), + "counters": [ + { + "id": counter.get("counterId"), + "value": counter.get("niceValue", {}).get("value"), + "initialValue": counter.get("niceValue", {}).get("initialValue"), + "unit": counter.get("niceValue", {}).get("type") + } + for counter in elem.get("counters", []) + ] + }) + return result + return None + except Exception: + return None \ No newline at end of file diff --git a/LidlConnect/api/user.py b/LidlConnect/api/user.py new file mode 100644 index 0000000..fc9ca84 --- /dev/null +++ b/LidlConnect/api/user.py @@ -0,0 +1,138 @@ +"""User-related API functionality for Lidl Connect.""" + +from typing import Dict, Any, Optional +from ..helpers import ttl_cache + +class UserDataMixin: + """User data methods for Lidl Connect API.""" + + USER_DATA_URL = "https://selfcare.lidl-connect.at/customer/dashboard/login-check" + + @ttl_cache(30) + def _get_user_data(self) -> Dict[str, Any]: + """ + Get user data from the server. + + Returns: + Dict: User data including name, type, accounts, etc. + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + } + + payload = {"userId": self.user_id, "endpointId": self.endpoint_id} + r = self.session.post(self.USER_DATA_URL, headers=headers, json=payload) + + if r.status_code != 200: + raise ValueError(f"User data request failed: {r.status_code} {r.text!r}") + + return r.json() + + @property + def user_name(self) -> Optional[str]: + """Get user's name.""" + try: + return self._get_user_data().get("name") + except Exception: + return None + + @property + def user_type(self) -> Optional[str]: + """Get user's type (e.g., 'CUSTOMER').""" + try: + return self._get_user_data().get("userType") + except Exception: + return None + + @property + def has_password(self) -> bool: + """Check if user has set a password.""" + try: + return self._get_user_data().get("hasPassword", False) + except Exception: + return False + + @property + def birth_date(self) -> Optional[str]: + """Get user's birth date.""" + try: + return self._get_user_data().get("birthDate") + except Exception: + return None + + @property + def status(self) -> Optional[str]: + """Get endpoint status (e.g., 'ACTIVE').""" + try: + data = self._get_user_data() + for account in data.get("accounts", []): + for endpoint in account.get("endpoints", []): + if endpoint.get("id") == self.endpoint_id: + return endpoint.get("status") + return None + except Exception: + return None + + @property + def customer_type(self) -> Optional[str]: + """Get customer type (e.g., 'ANONYM').""" + try: + return self._get_user_data().get("customerType") + except Exception: + return None + + @property + def customer_language(self) -> Optional[str]: + """Get customer language preference.""" + try: + return self._get_user_data().get("customerLanguage") + except Exception: + return None + + @property + def balance(self) -> Optional[float]: + """Get account balance.""" + try: + data = self._get_user_data() + for account in data.get("accounts", []): + for endpoint in account.get("endpoints", []): + if endpoint.get("id") == self.endpoint_id: + return endpoint.get("ocsBalance") + return None + except Exception: + return None + + @property + def activation_date(self) -> Optional[str]: + """Get activation date.""" + try: + data = self._get_user_data() + for account in data.get("accounts", []): + for endpoint in account.get("endpoints", []): + if endpoint.get("id") == self.endpoint_id: + return endpoint.get("activationDate") + return None + except Exception: + return None + + @property + def deactivation_date(self) -> Optional[str]: + """Get deactivation date.""" + try: + data = self._get_user_data() + for account in data.get("accounts", []): + for endpoint in account.get("endpoints", []): + if endpoint.get("id") == self.endpoint_id: + return endpoint.get("deactivationDate") + return None + except Exception: + return None \ No newline at end of file diff --git a/LidlConnect/api/utils.py b/LidlConnect/api/utils.py new file mode 100644 index 0000000..b5c8abb --- /dev/null +++ b/LidlConnect/api/utils.py @@ -0,0 +1,41 @@ +"""Utility functionality for Lidl Connect API.""" + +from typing import Dict, Any, Union + +class ApiMixin: + """General API utilities for Lidl Connect.""" + + def make_api_request(self, url: str, data: Dict = None, method: str = "POST") -> Union[Dict[str, Any], str]: + """ + Make a generic API request to Lidl Connect. + + Args: + url: API endpoint to call + data: Payload to send (optional) + method: HTTP method (default: POST) + + Returns: + Dict or str: API response (JSON parsed if Content-Type is application/json) + """ + if not self.logged_in or not self.csrf_token: + raise ValueError("Not logged in or missing CSRF token") + + headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=UTF-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": self.DASHBOARD_URL, + "X-CSRF-TOKEN": self.csrf_token, + "X-SELF-CARE": "1", + } + + if method.upper() == "POST": + r = self.session.post(url, headers=headers, json=data or {}) + else: + r = self.session.get(url, headers=headers) + + if r.status_code >= 400: + raise ValueError(f"API request failed: {r.status_code} {r.text!r}") + + return r.json() if r.headers.get('Content-Type') == 'application/json' else r.text \ No newline at end of file diff --git a/LidlConnect/auth.py b/LidlConnect/auth.py new file mode 100644 index 0000000..dcf13c7 --- /dev/null +++ b/LidlConnect/auth.py @@ -0,0 +1,56 @@ +"""Authentication-related functionality for Lidl Connect.""" + +from typing import Dict + +class AuthMixin: + """Authentication methods for Lidl Connect API.""" + + LOGIN_PUK_URL = "https://selfcare.lidl-connect.at/de/customer/login/puk" + LOGIN_PASSWORD_URL = "https://selfcare.lidl-connect.at/de/customer/login/account" + LOGOUT_URL = "https://selfcare.lidl-connect.at/de/customer/logout" + + @property + def login_url(self): + """Get the appropriate login URL based on credentials.""" + return self.LOGIN_PUK_URL if self.puk else self.LOGIN_PASSWORD_URL + + def login(self) -> bool: + """ + Log in to Lidl Connect. + + Returns: + bool: True if login successful + """ + login_headers = { + "Accept": "application/json", + "Content-Type": "application/json;charset=utf-8", + "X-Requested-With": "XMLHttpRequest", + "Origin": "https://selfcare.lidl-connect.at", + "Referer": "https://selfcare.lidl-connect.at/en/customer/login", + "X-AUTH-SELF-CARE": "1", + "locale": "en", + } + login_payload = {"identifier": self.identifier, "token": self.token} + + r = self.session.post(self.login_url, headers=login_headers, json=login_payload) + if r.status_code != 200: + return False + + self.logged_in = True + return True + + def logout(self) -> bool: + """ + Log out from Lidl Connect. + + Returns: + bool: True if logout successful, False otherwise + """ + if not self.logged_in: + return True + + r = self.session.get(self.LOGOUT_URL) + if r.status_code == 200 or r.status_code == 302: + self.logged_in = False + return True + return False \ No newline at end of file diff --git a/LidlConnect/client.py b/LidlConnect/client.py new file mode 100644 index 0000000..cea6d03 --- /dev/null +++ b/LidlConnect/client.py @@ -0,0 +1,84 @@ +"""Main client class for Lidl Connect API.""" + +import requests +import signal +import atexit + +from .auth import AuthMixin +from .extractors import ExtractorMixin +from .api.usage import UsageMixin +from .api.utils import ApiMixin +from .api.user import UserDataMixin +from .api.tariffs import TariffsMixin +from .api.invoices import InvoicesMixin + +class LidlConnect(AuthMixin, ExtractorMixin, UsageMixin, ApiMixin, UserDataMixin, TariffsMixin, InvoicesMixin): + """Client for interacting with Lidl Connect Self-Care portal.""" + + DASHBOARD_URL = "https://selfcare.lidl-connect.at/customer/dashboard/" + + def __init__(self, identifier: str, puk: str = None, password: str = None): + """ + Initialize Lidl Connect client. + + Args: + identifier: Your phone number or customer ID + puk: Your PUK code (optional if password provided) + password: Your password (optional if PUK provided) + """ + # Base components + self.identifier = identifier + self.puk = puk + self.password = password + self.token = puk if puk else password + + # Session and tokens + self.session = requests.Session() + self.csrf_token = None + self.endpoint_id = None + self.logged_in = False + + # User data + self.user_id = None + + atexit.register(self._cleanup) + signal.signal(signal.SIGINT, self._signal_handler) + signal.signal(signal.SIGTERM, self._signal_handler) + + def _cleanup(self): + """Clean up resources and log out when program exits.""" + if self.logged_in: + try: + self.logout() + except Exception as e: + print(f"Error during logout at program shutdown: {e}") + + def _signal_handler(self, signum, frame): + """Handle termination signals to ensure clean logout.""" + print("\nCaught termination signal. Logging out...") + self._cleanup() + signal.signal(signum, signal.SIG_DFL) + signal.raise_signal(signum) + + def initialize(self) -> bool: + """ + Initialize the client: login, fetch dashboard, and extract necessary tokens and IDs. + + Returns: + bool: True if initialization successful + """ + if not self.login(): + return False + + try: + soup = self._fetch_dashboard() + self.csrf_token = self._extract_csrf(soup) + self.user_id, self.endpoint_id = self._extract_user_and_endpoint(soup) + return True + except Exception as e: + print(f"Error during initialization: {e}") + return False + + def _fetch_dashboard(self): + """Fetch dashboard HTML and parse it.""" + return self._get_soup(self.DASHBOARD_URL) \ No newline at end of file diff --git a/LidlConnect/extractors.py b/LidlConnect/extractors.py new file mode 100644 index 0000000..8e56d50 --- /dev/null +++ b/LidlConnect/extractors.py @@ -0,0 +1,39 @@ +"""HTML and data extraction for Lidl Connect API.""" + +from bs4 import BeautifulSoup +import re +from typing import Tuple + +class ExtractorMixin: + """HTML extraction methods for Lidl Connect API.""" + + def _get_soup(self, url: str) -> BeautifulSoup: + """Get BeautifulSoup object from URL.""" + r = self.session.get(url) + return BeautifulSoup(r.text, "html.parser") + + def _extract_csrf(self, soup: BeautifulSoup) -> str: + """Extract CSRF token from dashboard HTML.""" + meta = soup.find("meta", {"name": "csrf-token"}) + if not meta or not meta.get("content"): + raise ValueError("CSRF token not found in dashboard HTML") + return meta["content"] + + def _extract_user_and_endpoint(self, soup: BeautifulSoup) -> Tuple[int, int]: + """Extract user ID and endpoint ID from dashboard HTML.""" + all_scripts = "" + for script in soup.find_all("script"): + if script.string: + all_scripts += script.string + + user_match = re.search(r"window\.user\s*=\s*\{.*?'user':\s*\{\s*\"id\":\s*(\d+).*?\"userType\":\s*\"CUSTOMER\"", all_scripts, re.DOTALL) + endpoint_match = re.search(r'"endpoints":\s*\[\{\s*"id":\s*(\d+)', all_scripts, re.DOTALL) + + if not user_match or not endpoint_match: + user_match = re.search(r'"id":\s*(\d+).*?"userType":\s*"CUSTOMER"', all_scripts, re.DOTALL) + endpoint_match = re.search(r'"endpoints":\s*\[\{\s*"id":\s*(\d+)', all_scripts, re.DOTALL) + + if not user_match or not endpoint_match: + raise ValueError("Could not extract userId or endpointId from scripts") + + return int(user_match.group(1)), int(endpoint_match.group(1)) \ No newline at end of file diff --git a/LidlConnect/helpers.py b/LidlConnect/helpers.py new file mode 100644 index 0000000..e0aa745 --- /dev/null +++ b/LidlConnect/helpers.py @@ -0,0 +1,34 @@ +"""Helper functions for Lidl Connect library.""" + +from functools import wraps +import time + +def ttl_cache(ttl_seconds=30): + """ + Time-based cache decorator. + + Args: + ttl_seconds: Number of seconds to cache the result + + Returns: + Decorated function with caching + """ + def decorator(func): + cache = {} + + @wraps(func) + def wrapper(*args, **kwargs): + now = time.time() + key = str(args) + str(kwargs) + + if key in cache: + result, timestamp = cache[key] + if now - timestamp < ttl_seconds: + return result + + result = func(*args, **kwargs) + cache[key] = (result, now) + return result + + return wrapper + return decorator \ No newline at end of file diff --git a/LidlConnect/requirements.txt b/LidlConnect/requirements.txt new file mode 100644 index 0000000..1f311f5 --- /dev/null +++ b/LidlConnect/requirements.txt @@ -0,0 +1,2 @@ +requests +bs4 \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..54d6f32 --- /dev/null +++ b/README.md @@ -0,0 +1,84 @@ +# LidlConnect.py + +[![PyPI version](https://badge.fury.io/py/LidlConnect.py.svg)](https://badge.fury.io/py/LidlConnect.py) +[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) + +A Python library for accessing your Lidl Connect account through the Self-Care API. + +## Features + +- 📱 View your current data, minutes, and SMS usage +- 💰 Check your account balance and payment history +- 📅 View tariff details and package validity dates +- 🧾 Access invoice and voucher history +- 🔄 Automatic session management with proper login/logout + +## Installation + +Install directly from PyPI: + +```bash +pip install LidlConnect.py +``` + +## Quick Start + +```python +from LidlConnect import LidlConnect + +# Initialize with PUK (preferred method) +client = LidlConnect(identifier="069012345678", puk="12345678") + +# Or initialize with password +# client = LidlConnect(identifier="069012345678", password="yourPassword") + +# Login and initialize connection +if not client.initialize(): + print("Failed to initialize client") + exit(1) + +# Get remaining data +data_info = client.get_remaining_data() +print(f"Data: {data_info['remaining']}/{data_info['total']} GiB") + +# Check minutes +minutes_info = client.get_remaining_minutes() +print(f"Minutes: {minutes_info['remaining']}/{minutes_info['total']} minutes") + +# Get EU roaming data +eu_data_info = client.get_remaining_eu_data() +print(f"EU Data: {eu_data_info['remaining']}/{eu_data_info['total']} GiB") + +# Access account information +print(f"User: {client.user_name}") +print(f"Balance: €{client.balance if client.balance is not None else 'N/A'}") +print(f"Package valid until: {client.tariff_package_valid_to}") + +# View payment history +client.print_invoices() + +# Logout when done (automatic on program exit, but explicit is better) +client.logout() +``` + +## Documentation + +For complete documentation of all features and methods, please refer to the [API documentation](apiDoc.md). + +## Requirements + +- Python 3.6+ +- `requests` library +- `beautifulsoup4` library + +## Supported Services + +This library currently supports the Austrian Lidl Connect service (https://selfcare.lidl-connect.at). It's designed to work with accounts that have at least one active tariff package. + +## License + +MIT License + +## Disclaimer + +This is an unofficial library and is not affiliated with, maintained, authorized, endorsed, or sponsored by Lidl or any of its affiliates (like Drei). diff --git a/apiDoc.md b/apiDoc.md new file mode 100644 index 0000000..385df86 --- /dev/null +++ b/apiDoc.md @@ -0,0 +1,535 @@ +# LidlConnect API Documentation + +A Python client library for interacting with the Lidl Connect Self-Care portal API. + +## Table of Contents + +- [Installation](#installation) +- [Quick Start](#quick-start) +- [Authentication](#authentication) +- [Core Client](#core-client) +- [Usage Data](#usage-data) +- [User Information](#user-information) +- [Tariffs](#tariffs) +- [Invoices and Payments](#invoices-and-payments) +- [Utility Methods](#utility-methods) +- [Caching](#caching) +- [Error Handling](#error-handling) + +## Installation + +```python +# Clone the repository +git clone https://github.com/MarcUs7i/LidlConnect.git + +# Install dependencies +pip install -r requirements.txt +``` + +Required dependencies: +- `requests` +- `beautifulsoup4` + +## Quick Start + +```python +from LidlConnect import LidlConnect + +# Initialize with PUK +client = LidlConnect(identifier="069012345678", puk="12345678") + +# Or initialize with password +# client = LidlConnect(identifier="069012345678", password="yourPassword") + +# Login and initialize connection +if not client.initialize(): + print("Failed to initialize client") + exit(1) + +# Get usage information +data_info = client.get_remaining_data() +print(f"Data: {data_info['remaining']}/{data_info['total']} GiB") + +# Logout when done +client.logout() +``` + +## Authentication + +### LidlConnect(identifier, puk=None, password=None) + +Initialize the Lidl Connect client. + +**Parameters:** +- `identifier` (str): Your phone number or customer ID +- `puk` (str, optional): Your PUK code (required if password not provided) +- `password` (str, optional): Your password (required if PUK not provided) + +**Example:** +```python +# Using PUK +client = LidlConnect(identifier="069012345678", puk="12345678") + +# Using password +client = LidlConnect(identifier="069012345678", password="yourPassword") +``` + +### initialize() → bool + +Initialize the client by logging in, fetching the dashboard, and extracting necessary tokens and IDs. + +**Returns:** +- `bool`: True if initialization successful, False otherwise + +**Example:** +```python +if not client.initialize(): + print("Failed to initialize client") + exit(1) +``` + +### login() → bool + +Log in to Lidl Connect using the provided credentials. + +**Returns:** +- `bool`: True if login successful, False otherwise + +**Example:** +```python +success = client.login() +``` + +### logout() → bool + +Log out from Lidl Connect, clearing the session. + +**Returns:** +- `bool`: True if logout successful, False otherwise + +**Example:** +```python +client.logout() +``` + +## Core Client + +The client automatically handles session management and token extraction, it tries to log out when the program is about to exit or to get killed. + +**Properties:** +- `identifier` (str): The phone number or customer ID +- `user_id` (int): The user ID extracted from the dashboard +- `endpoint_id` (int): The endpoint ID extracted from the dashboard +- `logged_in` (bool): Whether the client is currently logged in +- `csrf_token` (str): The CSRF token for API requests + +## Usage Data + +Methods for getting and displaying usage data. + +### get_usage_data() → Dict[str, Any] + +Get raw usage data for the current account. + +**Returns:** +- `Dict[str, Any]`: Usage data including instanceGroups with counters + +**Cache TTL:** 5 seconds + +**Example:** +```python +usage_data = client.get_usage_data() +``` + +### print_usage_summary(data=None) → None + +Pretty-print usage summary data. + +**Parameters:** +- `data` (Dict[str, Any], optional): Optional usage data. If None, will fetch new data + +**Example:** +```python +client.print_usage_summary() +``` + +### get_remaining_data() → Dict[str, float] + +Get remaining data balance (in GiB). + +**Returns:** +- `Dict[str, float]`: Dictionary with keys "remaining", "total", and "used" in GiB + +**Example:** +```python +data_info = client.get_remaining_data() +print(f"Data: {data_info['remaining']}/{data_info['total']} GiB") +``` + +### get_remaining_eu_data() → Dict[str, float] + +Get remaining EU data balance (in GiB). + +**Returns:** +- `Dict[str, float]`: Dictionary with keys "remaining", "total", and "used" in GiB + +**Example:** +```python +eu_data_info = client.get_remaining_eu_data() +print(f"EU Data: {eu_data_info['remaining']}/{eu_data_info['total']} GiB") +``` + +### get_remaining_minutes() → Dict[str, float] + +Get remaining voice minutes. + +**Returns:** +- `Dict[str, float]`: Dictionary with keys "remaining", "total", and "used" in minutes + +**Example:** +```python +minutes_info = client.get_remaining_minutes() +print(f"Minutes: {minutes_info['remaining']}/{minutes_info['total']} minutes") +``` + +### tariff_package_valid_from → Optional[str] + +Get the start date of the current tariff package. + +**Returns:** +- `str`: ISO formatted date string or None if not available + +**Example:** +```python +print(f"Package valid from: {client.tariff_package_valid_from}") +``` + +### tariff_package_valid_to → Optional[str] + +Get the end date of the current tariff package. + +**Returns:** +- `str`: ISO formatted date string or None if not available + +**Example:** +```python +print(f"Package valid to: {client.tariff_package_valid_to}") +``` + +### tariff_package_details → Optional[Dict[str, Any]] + +Get detailed information about the current tariff package. + +**Returns:** +- `Dict[str, Any]`: Dictionary containing name, category, validFrom, validTo and counter details or None if not available + +**Example:** +```python +details = client.tariff_package_details +if details: + print(f"Package name: {details['name']}") + print(f"Valid from: {details['validFrom']} to {details['validTo']}") +``` + +## User Information + +Methods for retrieving user account information. + +### _get_user_data() → Dict[str, Any] + +Get user data from the server. This is an internal method used by the properties below. + +**Returns:** +- `Dict[str, Any]`: User data including name, type, accounts, etc. + +**Cache TTL:** 30 seconds + +### user_name → Optional[str] + +Get user's name. + +**Returns:** +- `str`: User's name or None if not available + +**Example:** +```python +print(f"User name: {client.user_name}") +``` + +### user_type → Optional[str] + +Get user's type (e.g., 'CUSTOMER'). + +**Returns:** +- `str`: User type or None if not available + +**Example:** +```python +print(f"User type: {client.user_type}") +``` + +### has_password → bool + +Check if user has set a password. + +**Returns:** +- `bool`: True if the user has a password, False otherwise + +**Example:** +```python +if client.has_password: + print("User has a password set") +``` + +### birth_date → Optional[str] + +Get user's birth date. + +**Returns:** +- `str`: Birth date or None if not available + +**Example:** +```python +print(f"Birth date: {client.birth_date}") +``` + +### status → Optional[str] + +Get endpoint status (e.g., 'ACTIVE'). + +**Returns:** +- `str`: Endpoint status or None if not available + +**Example:** +```python +print(f"Status: {client.status}") +``` + +### customer_type → Optional[str] + +Get customer type (e.g., 'ANONYM'). + +**Returns:** +- `str`: Customer type or None if not available + +**Example:** +```python +print(f"Customer type: {client.customer_type}") +``` + +### customer_language → Optional[str] + +Get customer language preference. + +**Returns:** +- `str`: Customer language or None if not available + +**Example:** +```python +print(f"Language: {client.customer_language}") +``` + +### balance → Optional[float] + +Get account balance. + +**Returns:** +- `float`: Account balance or None if not available + +**Example:** +```python +print(f"Balance: €{client.balance if client.balance is not None else 'N/A'}") +``` + +### activation_date → Optional[str] + +Get activation date. + +**Returns:** +- `str`: ISO formatted activation date or None if not available + +**Example:** +```python +print(f"Activation date: {client.activation_date}") +``` + +### deactivation_date → Optional[str] + +Get deactivation date. + +**Returns:** +- `str`: ISO formatted deactivation date or None if not available + +**Example:** +```python +print(f"Deactivation date: {client.deactivation_date}") +``` + +## Tariffs + +Methods for retrieving and displaying available tariffs. + +### get_tariffs() → List[Dict[str, Any]] + +Get all available tariffs for the current account. + +**Returns:** +- `List[Dict[str, Any]]`: List of tariff objects with relevant information + +**Cache TTL:** 60 seconds + +**Example:** +```python +tariffs = client.get_tariffs() +``` + +### print_tariffs() → None + +Pretty-print available tariffs. + +**Example:** +```python +client.print_tariffs() +``` + +The output includes: +- Tariff name (German name if available) +- Tariff ID +- Details (cleaned from HTML, can look terrible) +- Featured status +- Visibility status + +## Invoices and Payments + +Methods for getting and displaying invoice and voucher history. + +### get_invoices() → List[Dict[str, Any]] + +Get list of invoices for the current account. + +**Returns:** +- `List[Dict[str, Any]]`: List of invoice objects with transaction details + +**Cache TTL:** 30 seconds + +**Example:** +```python +invoices = client.get_invoices() +``` + +### get_vouchers() → List[Dict[str, Any]] + +Get list of consumed vouchers for the current account. + +**Returns:** +- `List[Dict[str, Any]]`: List of voucher objects with transaction details + +**Cache TTL:** 30 seconds + +**Example:** +```python +vouchers = client.get_vouchers() +``` + +### print_invoices() → None + +Pretty-print invoice and voucher history. + +**Example:** +```python +client.print_invoices() +``` + +The output includes: +- Transaction details for invoices (ID, date, amount, payment method) +- Voucher details (ID, serial, value, consumption date) + +### get_total_spent() → float + +Calculate the total amount spent across all invoices and vouchers. + +**Returns:** +- `float`: Total amount in euros + +**Example:** +```python +total = client.get_total_spent() +print(f"Total spent: €{total:.2f}") +``` + +### last_payment_date → Optional[str] + +Get the date of the most recent payment from either invoices or vouchers. + +**Returns:** +- `str`: ISO formatted date string or None if no payments + +**Example:** +```python +print(f"Last payment date: {client.last_payment_date}") +``` + +## Utility Methods + +General API utilities for Lidl Connect. + +### make_api_request(url, data=None, method="POST") → Union[Dict[str, Any], str] + +Make a generic API request to Lidl Connect. + +**Parameters:** +- `url` (str): API endpoint to call +- `data` (Dict, optional): Payload to send +- `method` (str, optional): HTTP method (default: "POST") + +**Returns:** +- `Dict[str, Any]` or `str`: API response (JSON parsed if Content-Type is application/json) + +**Example:** +```python +response = client.make_api_request( + "https://selfcare.lidl-connect.at/customer/some-endpoint", + data={"key": "value"} +) +``` + +## Caching + +The library has a time-based cache for API requests to reduce load and improve performance. + +Key caching parameters: +- Usage data: 5 seconds TTL +- User data: 30 seconds TTL +- Invoices and vouchers: 30 seconds TTL +- Tariffs: 60 seconds TTL + +The cache is implemented using the `ttl_cache` decorator in helpers.py. + +## Error Handling + +Most methods raise `ValueError` exceptions when: +- Not logged in or missing CSRF token +- API requests fail +- Required data is missing + +All public properties safely handle exceptions internally and return None when errors occur, ensuring your application doesn't crash. + +Example of error handling: + +```python +try: + client = LidlConnect(identifier="069012345678", puk="12345678") + if not client.initialize(): + print("Failed to initialize client") + exit(1) + + # Your code here + +except ValueError as e: + print(f"API error: {e}") +except Exception as e: + print(f"Unexpected error: {e}") +finally: + if hasattr(client, 'logout') and client.logged_in: + client.logout() +``` \ No newline at end of file