Initial Commit

This commit is contained in:
MarcUs7i 2025-05-24 01:03:16 +02:00
commit a16b90e181
15 changed files with 1711 additions and 0 deletions

184
.gitignore vendored Normal file
View file

@ -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

5
LidlConnect/__init__.py Normal file
View file

@ -0,0 +1,5 @@
"""Lidl Connect API client library."""
from .client import LidlConnect
__all__ = ["LidlConnect"]

View file

@ -0,0 +1 @@
"""API modules for Lidl Connect."""

173
LidlConnect/api/invoices.py Normal file
View file

@ -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

138
LidlConnect/api/tariffs.py Normal file
View file

@ -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'<tr[^>]*>', '\n', teaser)
teaser = re.sub(r'<td[^>]*>', ' | ', teaser)
teaser = re.sub(r'</tr>', '', teaser)
teaser = re.sub(r'</td>', '', teaser)
# line breaks
teaser = teaser.replace('<br>', '\n').replace('<br/>', '\n')
teaser = teaser.replace('<p>', '').replace('</p>', '\n')
# divs and spans
teaser = re.sub(r'<div[^>]*>', '', teaser)
teaser = re.sub(r'</div>', '', teaser)
teaser = re.sub(r'<span[^>]*>', '', teaser)
teaser = re.sub(r'</span>', '', 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 ['<table>', '</table>', '<tbody>', '</tbody>']:
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()

197
LidlConnect/api/usage.py Normal file
View file

@ -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

138
LidlConnect/api/user.py Normal file
View file

@ -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

41
LidlConnect/api/utils.py Normal file
View file

@ -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

56
LidlConnect/auth.py Normal file
View file

@ -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

84
LidlConnect/client.py Normal file
View file

@ -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)

39
LidlConnect/extractors.py Normal file
View file

@ -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))

34
LidlConnect/helpers.py Normal file
View file

@ -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

View file

@ -0,0 +1,2 @@
requests
bs4

84
README.md Normal file
View file

@ -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).

535
apiDoc.md Normal file
View file

@ -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()
```