heynst/amazon_invoice_downloader
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add session-expiry email notifications with configurable SMTP

Browse source
This commit is contained in:
Stefan Heyn 2026-03-05 16:52:30 +01:00
parent 49e1f260c5
commit 36938a7f59
3 changed files with 128 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
README.md
View file

@ -36,6 +36,12 @@ Falls Amazon trotzdem Englisch zeigt, Sprache explizit per URL erzwingen:
python main.py configure --marketplace de --amazon-language de_DE --locale de-DE --timezone Europe/Berlin --currency EUR
```
Session-Ablauf per E-Mail melden (Default-Empfaenger ist `stefan.heyn@googlemail.com`):
```powershell
python main.py configure --notify-email stefan.heyn@googlemail.com --smtp-host smtp.gmail.com --smtp-port 587 --smtp-user "<dein-user>" --smtp-password "<app-passwort>"
```
Dann oeffnet sich ein Browser. Dort bei Amazon anmelden und auf Enter im Terminal druecken.
Die Session wird lokal gespeichert in:
@ -126,6 +132,11 @@ Optionen:
- `--debug`: zeigt, wie viele Detailseiten und Rechnungslinks gefunden werden
- `--debug-json [pfad]`: schreibt Laufdetails als JSON (ohne Pfad: Standarddatei)
- `configure --locale de-DE --timezone Europe/Berlin --amazon-language de_DE`: erzwingt deutsche Sprache (inkl. `language=de_DE`) und Berliner Zeitzone
- `configure --notify-email ... --smtp-host ...`: aktiviert E-Mail-Benachrichtigung bei Session-Ablauf
SMTP kann alternativ auch ueber Umgebungsvariablen gesetzt werden:
- `NOTIFY_EMAIL`, `SMTP_HOST`, `SMTP_PORT`, `SMTP_USER`, `SMTP_PASSWORD`, `SMTP_FROM`, `SMTP_STARTTLS`, `SMTP_SSL`
## Hinweise

8
docker-compose.yml
View file

@ -11,6 +11,14 @@
shm_size: "1gb"
environment:
- TZ=Europe/Berlin
- NOTIFY_EMAIL=${NOTIFY_EMAIL:-}
- SMTP_HOST=${SMTP_HOST:-}
- SMTP_PORT=${SMTP_PORT:-587}
- SMTP_USER=${SMTP_USER:-}
- SMTP_PASSWORD=${SMTP_PASSWORD:-}
- SMTP_FROM=${SMTP_FROM:-}
- SMTP_STARTTLS=${SMTP_STARTTLS:-true}
- SMTP_SSL=${SMTP_SSL:-false}
volumes:
- ./downloads:/downloads
- ./state:/root/.amazon_invoice_downloader

109
main.py
View file

@ -1,9 +1,12 @@
import argparse
import json
import os
import re
import smtplib
import time
from dataclasses import dataclass
from datetime import date, datetime
from email.message import EmailMessage
from pathlib import Path
from typing import Optional
from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse
@ -61,6 +64,53 @@ def build_context_options(config: dict) -> dict:
},
}
def strtobool(value: str) -> bool:
return value.strip().lower() in {"1", "true", "yes", "on"}
def get_notification_settings(config: dict) -> dict:
smtp_cfg = config.get("smtp", {})
return {
"recipient": os.getenv("NOTIFY_EMAIL", config.get("notify_email", "stefan.heyn@googlemail.com")),
"smtp_host": os.getenv("SMTP_HOST", smtp_cfg.get("host", "")),
"smtp_port": int(os.getenv("SMTP_PORT", str(smtp_cfg.get("port", 587)))),
"smtp_user": os.getenv("SMTP_USER", smtp_cfg.get("user", "")),
"smtp_password": os.getenv("SMTP_PASSWORD", smtp_cfg.get("password", "")),
"smtp_from": os.getenv("SMTP_FROM", smtp_cfg.get("from_addr", "")),
"smtp_starttls": strtobool(os.getenv("SMTP_STARTTLS", str(smtp_cfg.get("starttls", True)))),
"smtp_ssl": strtobool(os.getenv("SMTP_SSL", str(smtp_cfg.get("ssl", False)))),
}
def send_notification(config: dict, subject: str, body: str) -> None:
settings = get_notification_settings(config)
recipient = settings["recipient"]
host = settings["smtp_host"]
if not host or not recipient:
return
sender = settings["smtp_from"] or settings["smtp_user"] or recipient
msg = EmailMessage()
msg["Subject"] = subject
msg["From"] = sender
msg["To"] = recipient
msg.set_content(body)
smtp_cls = smtplib.SMTP_SSL if settings["smtp_ssl"] else smtplib.SMTP
with smtp_cls(host, settings["smtp_port"], timeout=20) as smtp:
if not settings["smtp_ssl"] and settings["smtp_starttls"]:
smtp.starttls()
if settings["smtp_user"]:
smtp.login(settings["smtp_user"], settings["smtp_password"])
smtp.send_message(msg)
def is_login_page(page) -> bool:
url = page.url.lower()
if any(part in url for part in ["/ap/signin", "/signin", "openid.oa"]):
return True
email_fields = page.locator('input[type="email"], input[name="email"], #ap_email')
return email_fields.count() > 0
def parse_iso_date(value: str) -> date:
try:
@ -303,6 +353,16 @@ def configure(args) -> None:
"timezone": args.timezone,
"currency": args.currency,
"amazon_language": args.amazon_language,
"notify_email": args.notify_email,
"smtp": {
"host": args.smtp_host,
"port": args.smtp_port,
"user": args.smtp_user,
"password": args.smtp_password,
"from_addr": args.smtp_from,
"starttls": not args.smtp_no_starttls,
"ssl": args.smtp_ssl,
},
}
save_config(config)
ensure_app_dir()
@ -350,6 +410,7 @@ def download(args) -> None:
download_dir = Path(args.output or config["download_dir"]).expanduser().resolve()
download_dir.mkdir(parents=True, exist_ok=True)
debug_json_target = args.debug_json or (str(DEFAULT_DEBUG_JSON_PATH) if args.debug else None)
recipient = get_notification_settings(config).get("recipient", "")
with sync_playwright() as p:
browser = p.chromium.launch(headless=args.headless if args.headless is not None else bool(config.get("headless", True)))
@ -378,6 +439,25 @@ def download(args) -> None:
if args.debug:
print(f"[debug] Wechsle auf Jahresfilter {year}: {filtered_url}")
page.goto(filtered_url, wait_until="domcontentloaded", timeout=15000)
if is_login_page(page):
msg = (
"Amazon-Session ist abgelaufen oder Login wurde angefordert.\n"
f"URL: {page.url}\n"
f"Zeitraum: {start_date.isoformat()} bis {end_date.isoformat()}\n"
"Bitte 'configure' erneut ausfuehren."
)
try:
send_notification(
config,
subject="Amazon Invoice Downloader: Session abgelaufen",
body=msg,
)
except Exception as notify_exc:
print(f"[warn] E-Mail-Benachrichtigung fehlgeschlagen: {notify_exc}")
raise SystemExit(
"Session abgelaufen. Bitte 'configure' erneut ausfuehren."
+ (f" Benachrichtigung an {recipient} gesendet." if recipient else "")
)
visited_page_urls = set()
for page_idx in range(args.max_pages):
@ -433,6 +513,25 @@ def download(args) -> None:
wait_until="domcontentloaded",
timeout=15000,
)
if is_login_page(page):
msg = (
"Amazon-Session ist waehrend der Pagination abgelaufen.\n"
f"URL: {page.url}\n"
f"Zeitraum: {start_date.isoformat()} bis {end_date.isoformat()}\n"
"Bitte 'configure' erneut ausfuehren."
)
try:
send_notification(
config,
subject="Amazon Invoice Downloader: Session waehrend Download abgelaufen",
body=msg,
)
except Exception as notify_exc:
print(f"[warn] E-Mail-Benachrichtigung fehlgeschlagen: {notify_exc}")
raise SystemExit(
"Session abgelaufen. Bitte 'configure' erneut ausfuehren."
+ (f" Benachrichtigung an {recipient} gesendet." if recipient else "")
)
except PlaywrightTimeoutError:
break
@ -516,6 +615,14 @@ def build_parser() -> argparse.ArgumentParser:
p_config.add_argument("--timezone", default="Europe/Berlin", help="Zeitzone, z. B. Europe/Berlin")
p_config.add_argument("--currency", default="EUR", help="Waehrungshinweis fuer Konfiguration")
p_config.add_argument("--amazon-language", default="de_DE", help="Amazon URL-Sprache, z. B. de_DE")
p_config.add_argument("--notify-email", default="stefan.heyn@googlemail.com", help="Empfaenger fuer Ablauf-Benachrichtigungen")
p_config.add_argument("--smtp-host", default="", help="SMTP-Server, z. B. smtp.gmail.com")
p_config.add_argument("--smtp-port", type=int, default=587, help="SMTP-Port, Standard 587")
p_config.add_argument("--smtp-user", default="", help="SMTP-Benutzer")
p_config.add_argument("--smtp-password", default="", help="SMTP-Passwort oder App-Passwort")
p_config.add_argument("--smtp-from", default="", help="Absenderadresse (optional)")
p_config.add_argument("--smtp-ssl", action="store_true", help="SMTP ueber SSL (typisch Port 465)")
p_config.add_argument("--smtp-no-starttls", action="store_true", help="STARTTLS deaktivieren")
p_config.add_argument(
"--login-wait-seconds",
type=int,
@ -551,3 +658,5 @@ def main() -> None:
if __name__ == "__main__":
main()