File: //opt/BulkKindleUSBDownloader/bookp.py
#!/usr/bin/env python3
"""
Fully überarbeitetes Bulk-Kindle-Downloader-Skript
-------------------------------------------------
• Robuste Selenium-Initialisierung mit webdriver-manager
• Headless-Support via --showbrowser / pyvirtualdisplay
• Generischer JSON-Parser, der OwnershipData/ content/ records usw. erkennt
• Einige kleinere Verbesserungen (Logging, Typ-Hints, F-Strings)
Stand: Juli 2025
"""
import getpass
import json
import logging
import os
import re
import shutil
import sys
import tempfile
import urllib.parse
from argparse import ArgumentParser
from pathlib import Path
from time import sleep
from typing import Dict, List, Tuple
import requests
from pyvirtualdisplay import Display
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.common.by import By
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.support.ui import WebDriverWait
from webdriver_manager.chrome import ChromeDriverManager
# --------------------------------------------------------------------------- #
# Grundeinstellungen
# --------------------------------------------------------------------------- #
USER_AGENT_HEADER = {"User-Agent": "krumpli"}
logger = logging.getLogger(__name__)
ENVIRONMENTS: Dict[str, Dict[str, str]] = {
"UK": {"base_url": "https://www.amazon.co.uk", "account_list_selector": "nav-link-accountList",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email_login"},
"USA": {"base_url": "https://www.amazon.com", "account_list_selector": "nav-link-accountList-nav-line-1",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email"},
"Germany": {"base_url": "https://www.amazon.de", "account_list_selector": "nav-link-accountList",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email"},
"Italy": {"base_url": "https://www.amazon.it", "account_list_selector": "nav-link-accountList-nav-line-1",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email"},
"Canada": {"base_url": "https://www.amazon.ca", "account_list_selector": "nav-link-accountList",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email_login"},
"France": {"base_url": "https://www.amazon.fr", "account_list_selector": "nav-link-accountList",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email"},
"Japan": {"base_url": "https://www.amazon.co.jp", "account_list_selector": "nav-link-accountList",
"sign_in_selector": "#nav-flyout-ya-signin > a.nav-action-signin-button", "email_selector": "ap_email"},
}
# --------------------------------------------------------------------------- #
# Selenium-Hilfsfunktionen
# --------------------------------------------------------------------------- #
def _init_browser(headless: bool, proxy: str | None) -> webdriver.Chrome:
"""Erzeugt einen Chrome-WebDriver mit webdriver-manager – stets passender
chromedriver-Version."""
chrome_opts = Options()
chrome_opts.add_argument("--window-size=1920,1080")
chrome_opts.add_argument("--disable-gpu")
chrome_opts.add_argument("--no-sandbox")
if headless:
chrome_opts.add_argument("--headless=new")
# Proxy?
if proxy:
chrome_opts.add_argument(f"--proxy-server={proxy}")
# Versuche, das Chrome-Binary zu lokalisieren (Snap-Bug umgehen)
for candidate in ("/usr/bin/google-chrome", "/usr/bin/chromium-browser", shutil.which("google-chrome")):
if candidate and Path(candidate).exists():
chrome_opts.binary_location = candidate
break
# Passenden Driver installieren/finden
service = Service(ChromeDriverManager().install())
return webdriver.Chrome(service=service, options=chrome_opts)
def _start_virtual_display() -> Display:
"""Starte ein unsichtbares X-Display für Headless-Selenium."""
display = Display(visible=0)
display.start()
return display
# --------------------------------------------------------------------------- #
# Amazon-Login & Sitzungsaufbau
# --------------------------------------------------------------------------- #
def create_session(email: str, password: str, oath: str, env: Dict[str, str],
browser_visible: bool, proxy: str | None
) -> Tuple[Dict[str, str], str, str]:
"""Loggt den User ein und extrahiert Cookies, CSRF-Token & Customer-ID."""
display = _start_virtual_display() if not browser_visible else None
browser = _init_browser(headless=not browser_visible, proxy=proxy)
base_url = env["base_url"]
logger.info("⇢ Öffne Amazon-Startseite")
browser.get(base_url)
# Hover + Sign-In-Click
logger.info("⇢ Sign-In")
WebDriverWait(browser, 8).until(
EC.presence_of_element_located((By.ID, env["account_list_selector"]))
)
ActionChains(browser).move_to_element(
browser.find_element(By.ID, env["account_list_selector"])
).perform()
browser.find_element(By.CSS_SELECTOR, env["sign_in_selector"]).click()
# E-Mail + Passwort
WebDriverWait(browser, 8).until(
EC.presence_of_element_located((By.ID, env["email_selector"]))
)
browser.find_element(By.ID, env["email_selector"]).send_keys(email)
browser.find_element(By.ID, "continue").click()
WebDriverWait(browser, 8).until(
EC.presence_of_element_located((By.ID, "ap_password"))
)
browser.find_element(By.ID, "ap_password").send_keys(password)
browser.find_element(By.ID, "signInSubmit").click()
# MFA?
if oath:
WebDriverWait(browser, 8).until(
EC.presence_of_element_located((By.ID, "auth-mfa-otpcode"))
)
browser.find_element(By.ID, "auth-mfa-otpcode").send_keys(oath)
browser.find_element(By.ID, "auth-signin-button").click()
sleep(4) # Amazon braucht manchmal einen Moment
# Content-Seite aufrufen, um Tokens im HTML zu finden
logger.info("⇢ Hole CSRF-Token & Customer-ID")
browser.get(f"{base_url}/hz/mycd/digital-console/contentlist/booksAll/dateDsc/")
page_source = browser.page_source
csrf_token = re.search(r'var csrfToken = "(.*?)";', page_source)
custid = re.search(r'customerId: "(.*?)"', page_source)
csrf_token = csrf_token.group(1) if csrf_token else ""
custid = custid.group(1) if custid else ""
cookies = {c["name"]: c["value"] for c in browser.get_cookies()}
browser.quit()
if display:
display.stop()
return cookies, csrf_token, custid
# --------------------------------------------------------------------------- #
# API-Helpers
# --------------------------------------------------------------------------- #
def _extract_items(rr: dict) -> Tuple[List[dict], bool]:
"""Durchsucht Amazons Antwort nach einem Block, der 'items' + 'hasMoreItems'
enthält – egal ob der Ober-Key 'OwnershipData', 'content', 'records' o. Ä.
heißt."""
for block in rr.values():
if isinstance(block, dict) and "items" in block:
items = block["items"]
has_more = block.get("hasMoreItems", False)
return items, has_more
raise KeyError("Kein items-Block gefunden! Antwort:\n" + json.dumps(rr)[:1000])
# --------------------------------------------------------------------------- #
# Datenabfragen
# --------------------------------------------------------------------------- #
def get_devices(cookies: Dict[str, str], csrf_token: str, env: Dict[str, str]) -> List[dict]:
logger.info("⇢ Hole Geräte")
data_json = {"param": {"GetDevices": {}}}
r = requests.post(f"{env['base_url']}/hz/mycd/ajax",
data={"data": json.dumps(data_json), "csrfToken": csrf_token},
headers=USER_AGENT_HEADER, cookies=cookies)
return [d for d in r.json()["GetDevices"]["devices"] if "deviceSerialNumber" in d]
def get_asins(cookies: Dict[str, str], csrf_token: str, env: Dict[str, str]) -> List[str]:
logger.info("⇢ Hole eBook-Liste (paginierend)")
start, batch = 0, 100
asins: List[str] = []
while True:
data_json = {"param": {"OwnershipData": {
"sortOrder": "DESCENDING", "sortIndex": "DATE",
"startIndex": start, "batchSize": batch,
"contentType": "Ebook", "itemStatus": ["Active"],
"originType": ["Prime", "Purchase", "Sharing"]}}}
r = requests.post(f"{env['base_url']}/hz/mycd/ajax",
data={"data": json.dumps(data_json), "csrfToken": csrf_token},
headers=USER_AGENT_HEADER, cookies=cookies)
items, has_more = _extract_items(r.json())
asins.extend(book["asin"] for book in items)
if not has_more:
break
start += batch
logger.info(f" → {len(asins)} Bücher gefunden")
return asins
def download_books(cookies: Dict[str, str], device: dict,
asins: List[str], custid: str, directory: Path) -> None:
logger.info(f"⇢ Starte Download von {len(asins)} Büchern")
cdn_url = "https://cde-ta-g7g.amazon.com/FionaCDEServiceEngine/FSDownloadContent"
params_tpl = "type=EBOK&key={asin}&fsn={fsn}&device_type={dtype}&customerId={cid}&authPool=Amazon"
for asin in asins:
try:
params = params_tpl.format(
asin=asin, fsn=device["deviceSerialNumber"],
dtype=device["deviceType"], cid=custid)
r = requests.get(cdn_url, params=params, headers=USER_AGENT_HEADER,
cookies=cookies, stream=True, timeout=60)
filename_enc = re.search(r"filename\*=UTF-8''(.+)", r.headers["Content-Disposition"]).group(1)
filename = urllib.parse.unquote(filename_enc).replace("/", "_")
target = directory / filename
with target.open("wb") as f:
for chunk in r.iter_content(chunk_size=8192):
f.write(chunk)
logger.info(f" ✔ {asin}: {filename}")
except Exception as e:
logger.error(f" ✘ {asin}: {e}")
# --------------------------------------------------------------------------- #
# CLI-Entry-Point
# --------------------------------------------------------------------------- #
def main() -> int:
parser = ArgumentParser(description="Amazon-E-Book-Downloader")
parser.add_argument("--verbose", action="store_true", help="zeige Info-Logs")
parser.add_argument("--showbrowser", action="store_true", help="Browserfenster anzeigen")
parser.add_argument("--email", required=True, help="Amazon-E-Mail")
parser.add_argument("--password", help="Amazon-Passwort")
parser.add_argument("--oath", help="MFA-Code (falls aktiviert)")
parser.add_argument("--outputdir", default="books", help="Download-Ordner")
parser.add_argument("--proxy", help="HTTP-Proxy z. B. http://127.0.0.1:8080")
parser.add_argument("--asin", nargs="*", help="ASIN-Liste (optional)")
parser.add_argument("--logfile", help="Log-Datei")
args = parser.parse_args()
# Logging
logger.setLevel(logging.INFO if args.verbose else logging.WARNING)
fmt = logging.Formatter("[%(levelname)s] %(asctime)s %(message)s", "%H:%M:%S")
handler = logging.StreamHandler()
handler.setFormatter(fmt)
logger.addHandler(handler)
if args.logfile:
fh = logging.FileHandler(args.logfile)
fh.setFormatter(fmt)
logger.addHandler(fh)
# Credentials
pwd = args.password or getpass.getpass("Amazon-Passwort: ")
mfa = args.oath or getpass.getpass("MFA-Code (Enter wenn keiner): ")
# Länderauswahl
keys = list(ENVIRONMENTS)
print("\nLand auswählen:")
for idx, key in enumerate(keys):
print(f" {idx}. {key}")
while True:
try:
env = ENVIRONMENTS[keys[int(input('Nummer: '))]]
break
except (ValueError, KeyError, IndexError):
print("Ungültige Eingabe – bitte Zahl wiederholen.")
# Zielordner
out_dir = Path(args.outputdir).expanduser()
out_dir.mkdir(parents=True, exist_ok=True)
# Session
cookies, csrf, custid = create_session(
email=args.email, password=pwd, oath=mfa,
env=env, browser_visible=args.showbrowser, proxy=args.proxy)
# ASINs sammeln
asins = args.asin or get_asins(cookies, csrf, env)
# Gerät wählen
devices = get_devices(cookies, csrf, env)
print("\nGerät auswählen:")
for idx, d in enumerate(devices):
print(f" {idx}. {d['deviceAccountName']}")
while True:
try:
device = devices[int(input('Nummer: '))]
break
except (ValueError, IndexError):
print("Ungültige Eingabe – bitte Zahl wiederholen.")
# Download
download_books(cookies, device, asins, custid, out_dir)
sn = device["deviceSerialNumber"]
print(f"\nFertig! DeDRM-Seriennummer: {sn}\n"
"→ https://github.com/noDRM/DeDRM_tools")
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
logger.info("Abbruch durch Benutzer")