Czechman
/
humbleParser
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fehlerkorrektur und debugausgabe

This commit is contained in:
Czechman 2025-02-16 13:58:59 +01:00
parent 60964f7f6e
commit 19a01a4f90
1 changed files with 62 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
bundle_checker.py
View File

@ -2,28 +2,41 @@
import requests import requests
import json import json
import hashlib import hashlib
import logging
from datetime import datetime from datetime import datetime
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
# Importiere declarative_base aus sqlalchemy.orm (SQLAlchemy 2.0-konform)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, Text from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey, Text
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
# Basis-Klasse für SQLAlchemy-Modelle # Konfiguriere Logging – passe das Level bei Bedarf an (DEBUG, INFO, WARNING, ERROR)
DEBUG_LEVEL = logging.DEBUG
logging.basicConfig(
level=DEBUG_LEVEL,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)
# Basis-Klasse für SQLAlchemy-Modelle (SQLAlchemy 2.0-konform)
Base = declarative_base() Base = declarative_base()
# Tabelle für das Bundle (statische Identifikation) # Tabelle für das Bundle (statische Identifikation)
class Bundle(Base): class Bundle(Base):
__tablename__ = 'bundles' __tablename__ = 'bundles'
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
machine_name = Column(String, unique=True) # z. B. "linuxfrombeginnertoprofessionaloreilly_bookbundle" machine_name = Column(String, unique=True)
human_name = Column(String) human_name = Column(String)
# current_version_id verweist auf die aktuelle Version in bundle_versions
current_version_id = Column(Integer, ForeignKey('bundle_versions.id')) current_version_id = Column(Integer, ForeignKey('bundle_versions.id'))
# Beziehung zur aktuellen Version # Beziehung zur aktuellen Version; post_update=True hilft bei zirkulären Abhängigkeiten
current_version = relationship("BundleVersion", uselist=False, foreign_keys=[current_version_id]) current_version = relationship("BundleVersion", uselist=False,
foreign_keys=[current_version_id],
post_update=True)
# Alle Versionen (historisch) # Alle Versionen (historisch)
versions = relationship("BundleVersion", back_populates="bundle", foreign_keys='BundleVersion.bundle_id') versions = relationship("BundleVersion", back_populates="bundle",
foreign_keys=lambda: [BundleVersion.bundle_id])
# Verkaufshistorie # Verkaufshistorie
sales_history = relationship("BundleSalesHistory", back_populates="bundle") sales_history = relationship("BundleSalesHistory", back_populates="bundle")
@ -32,13 +45,14 @@ class BundleVersion(Base):
__tablename__ = 'bundle_versions' __tablename__ = 'bundle_versions'
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
bundle_id = Column(Integer, ForeignKey('bundles.id')) bundle_id = Column(Integer, ForeignKey('bundles.id'))
version_hash = Column(String) # SHA-256 Hash der relevanten Daten version_hash = Column(String)
version_data = Column(Text) # Alle relevanten Bundle-Daten als JSON-String version_data = Column(Text) # Relevante Bundle-Daten als JSON-String
timestamp = Column(DateTime, default=datetime.utcnow) timestamp = Column(DateTime, default=datetime.utcnow)
bundle = relationship("Bundle", back_populates="versions") # Eindeutige Beziehung: wir verwenden hier explizit bundle_id
bundle = relationship("Bundle", back_populates="versions", foreign_keys=[bundle_id])
# Tabelle für Verkaufshistorie (zur zeitlichen Analyse der Verkaufszahlen) # Tabelle für Verkaufshistorie
class BundleSalesHistory(Base): class BundleSalesHistory(Base):
__tablename__ = 'bundle_sales_history' __tablename__ = 'bundle_sales_history'
id = Column(Integer, primary_key=True) id = Column(Integer, primary_key=True)
@ -51,17 +65,23 @@ class BundleSalesHistory(Base):
def calculate_hash(data: dict) -> str: def calculate_hash(data: dict) -> str:
"""Berechnet einen SHA-256 Hash aus dem sortierten JSON-String der relevanten Daten.""" """Berechnet einen SHA-256 Hash aus dem sortierten JSON-String der relevanten Daten."""
json_string = json.dumps(data, sort_keys=True, ensure_ascii=False) json_string = json.dumps(data, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(json_string.encode('utf-8')).hexdigest() hash_value = hashlib.sha256(json_string.encode('utf-8')).hexdigest()
logger.debug(f"Berechneter Hash: {hash_value}")
return hash_value
def fetch_bundle_data(url: str) -> dict: def fetch_bundle_data(url: str) -> dict:
"""Lädt die Detailseite eines Bundles und extrahiert den JSON-Inhalt aus dem <script>-Tag 'webpack-bundle-page-data'.""" """Lädt die Detailseite eines Bundles und extrahiert den JSON-Inhalt aus dem <script>-Tag 'webpack-bundle-page-data'."""
logger.info(f"Rufe Bundle-Daten von {url} ab...")
response = requests.get(url) response = requests.get(url)
response.raise_for_status() response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser") soup = BeautifulSoup(response.text, "html.parser")
script_tag = soup.find("script", {"id": "webpack-bundle-page-data", "type": "application/json"}) script_tag = soup.find("script", {"id": "webpack-bundle-page-data", "type": "application/json"})
if not script_tag: if not script_tag:
raise ValueError("Kein JSON-Datenblock 'webpack-bundle-page-data' gefunden auf der Detailseite!") logger.error("Kein JSON-Datenblock 'webpack-bundle-page-data' gefunden!")
return json.loads(script_tag.string) raise ValueError("Kein JSON-Datenblock gefunden!")
data = json.loads(script_tag.string)
logger.debug(f"Erhaltener JSON-Block: {str(data)[:200]} ...")
return data
def process_bundle(session, url: str): def process_bundle(session, url: str):
""" """
@ -74,31 +94,36 @@ def process_bundle(session, url: str):
try: try:
data = fetch_bundle_data(url) data = fetch_bundle_data(url)
except Exception as e: except Exception as e:
print(f"Fehler beim Laden der Bundle-Daten von {url}: {e}") logger.error(f"Fehler beim Laden der Bundle-Daten von {url}: {e}")
return return
bundle_data = data.get("bundleData", {}) bundle_data = data.get("bundleData", {})
machine_name = bundle_data.get("machine_name", "") machine_name = bundle_data.get("machine_name", "")
human_name = bundle_data.get("human_name", "") human_name = bundle_data.get("human_name", "")
logger.info(f"Verarbeite Bundle '{human_name}' (machine_name: {machine_name})")
# Relevanter Datenausschnitt – hier nehmen wir das gesamte bundleData # Wir nehmen das gesamte bundleData als relevanten Datenausschnitt
relevant_data = bundle_data relevant_data = bundle_data
new_hash = calculate_hash(relevant_data) new_hash = calculate_hash(relevant_data)
# Suche, ob das Bundle bereits existiert (über machine_name) # Suche, ob das Bundle bereits existiert
bundle = session.query(Bundle).filter_by(machine_name=machine_name).first() bundle = session.query(Bundle).filter_by(machine_name=machine_name).first()
if not bundle: if not bundle:
logger.info(f"Neues Bundle gefunden: {human_name}")
bundle = Bundle(machine_name=machine_name, human_name=human_name) bundle = Bundle(machine_name=machine_name, human_name=human_name)
session.add(bundle) session.add(bundle)
session.commit() session.commit()
else:
logger.debug(f"Bundle '{human_name}' existiert bereits (ID: {bundle.id})")
# Hole die aktuellste Version dieses Bundles (nach timestamp sortiert) # Hole die aktuellste Version dieses Bundles
latest_version = (session.query(BundleVersion) latest_version = (session.query(BundleVersion)
.filter_by(bundle_id=bundle.id) .filter_by(bundle_id=bundle.id)
.order_by(BundleVersion.timestamp.desc()) .order_by(BundleVersion.timestamp.desc())
.first()) .first())
if latest_version is None or latest_version.version_hash != new_hash: if latest_version is None or latest_version.version_hash != new_hash:
logger.info(f"Neue Version für Bundle '{human_name}' wird gespeichert (alter Hash: {latest_version.version_hash if latest_version else 'None'}, neuer Hash: {new_hash}).")
new_version = BundleVersion( new_version = BundleVersion(
bundle_id=bundle.id, bundle_id=bundle.id,
version_hash=new_hash, version_hash=new_hash,
@ -108,12 +133,10 @@ def process_bundle(session, url: str):
session.commit() session.commit()
bundle.current_version_id = new_version.id bundle.current_version_id = new_version.id
session.commit() session.commit()
print(f"Neue Version für Bundle '{human_name}' gespeichert.")
else: else:
print(f"Bundle '{human_name}' hat sich nicht geändert.") logger.info(f"Bundle '{human_name}' hat sich nicht geändert.")
# Verkaufszahlen extrahieren – hier wird angenommen, dass sie entweder direkt in bundleData # Verkaufszahlen extrahieren – angenommen, sie stehen entweder direkt in bundleData oder unter basic_data
# oder unter basic_data stehen, im Feld "bundles_sold|decimal"
bundles_sold = bundle_data.get("bundles_sold|decimal") bundles_sold = bundle_data.get("bundles_sold|decimal")
if bundles_sold is None: if bundles_sold is None:
bundles_sold = bundle_data.get("basic_data", {}).get("bundles_sold|decimal") bundles_sold = bundle_data.get("basic_data", {}).get("bundles_sold|decimal")
@ -126,16 +149,17 @@ def process_bundle(session, url: str):
) )
session.add(sales_entry) session.add(sales_entry)
session.commit() session.commit()
print(f"Verkaufszahlen für Bundle '{human_name}' aktualisiert: {sales_value}") logger.info(f"Verkaufszahlen für Bundle '{human_name}' aktualisiert: {sales_value}")
except Exception as e: except Exception as e:
print(f"Fehler beim Speichern der Verkaufszahlen für {human_name}: {e}") logger.error(f"Fehler beim Speichern der Verkaufszahlen für {human_name}: {e}")
else:
logger.debug(f"Keine Verkaufszahlen für Bundle '{human_name}' gefunden.")
def get_bundle_urls(overview_url: str) -> list: def get_bundle_urls(overview_url: str) -> list:
""" """
Ruft die Übersichtsseite ab und extrahiert alle Bundle-URLs aus dem JSON-Datenblock Ruft die Übersichtsseite ab und extrahiert alle Bundle-URLs aus dem JSON-Datenblock im <script>-Tag 'landingPage-json-data'.
im <script>-Tag mit der ID "landingPage-json-data". Dies ist zuverlässig, da die
Bundles dort dynamisch eingebunden werden.
""" """
logger.info(f"Rufe Übersichtsseite {overview_url} ab...")
response = requests.get(overview_url) response = requests.get(overview_url)
response.raise_for_status() response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser") soup = BeautifulSoup(response.text, "html.parser")
@ -143,46 +167,44 @@ def get_bundle_urls(overview_url: str) -> list:
landing_script = soup.find("script", {"id": "landingPage-json-data", "type": "application/json"}) landing_script = soup.find("script", {"id": "landingPage-json-data", "type": "application/json"})
if landing_script: if landing_script:
landing_data = json.loads(landing_script.string) landing_data = json.loads(landing_script.string)
# Wir erwarten, dass in landing_data["data"] die Kategorien "books", "games", "software" enthalten sind. logger.debug(f"Landing Page JSON (gekürzt): {str(landing_data)[:200]} ...")
# Durchlaufe die Kategorien "books", "games" und "software"
for category in ["books", "games", "software"]: for category in ["books", "games", "software"]:
cat_data = landing_data.get("data", {}).get(category, {}) cat_data = landing_data.get("data", {}).get(category, {})
for section in cat_data.get("mosaic", []): for section in cat_data.get("mosaic", []):
for product in section.get("products", []): for product in section.get("products", []):
# Annahme: Der Link steht im Feld "product_url"
url = product.get("product_url", "") url = product.get("product_url", "")
if url: if url:
full_url = requests.compat.urljoin(overview_url, url) full_url = requests.compat.urljoin(overview_url, url)
bundle_urls.append(full_url) bundle_urls.append(full_url)
else: else:
print("Kein JSON-Datenblock 'landingPage-json-data' auf der Übersichtsseite gefunden.") logger.warning("Kein JSON-Datenblock 'landingPage-json-data' auf der Übersichtsseite gefunden.")
logger.info(f"Gefundene Bundle-URLs: {len(bundle_urls)}")
return bundle_urls return bundle_urls
def main(): def main():
# Datenbank-Verbindung (hier SQLite)
engine = create_engine('sqlite:///bundles.db') engine = create_engine('sqlite:///bundles.db')
Base.metadata.create_all(engine) Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine) Session = sessionmaker(bind=engine)
session = Session() session = Session()
# URL der Bundles-Übersichtsseite
overview_url = "https://www.humblebundle.com/bundles" overview_url = "https://www.humblebundle.com/bundles"
try: try:
bundle_urls = get_bundle_urls(overview_url) bundle_urls = get_bundle_urls(overview_url)
if not bundle_urls: if not bundle_urls:
print("Keine Bundle-URLs gefunden! Überprüfe den JSON-Datenblock oder den Selektor in get_bundle_urls().") logger.error("Keine Bundle-URLs gefunden! Überprüfe den JSON-Datenblock oder den Selektor in get_bundle_urls().")
return return
except Exception as e: except Exception as e:
print(f"Fehler beim Abrufen der Übersichtsseite: {e}") logger.error(f"Fehler beim Abrufen der Übersichtsseite: {e}")
return return
# Verarbeite alle gefundenen Bundle-URLs # Für jede Bundle-URL verarbeiten
for url in bundle_urls: for url in bundle_urls:
print(f"Verarbeite Bundle: {url}") logger.info(f"Verarbeite Bundle: {url}")
try: try:
process_bundle(session, url) process_bundle(session, url)
except Exception as e: except Exception as e:
print(f"Fehler beim Verarbeiten von {url}: {e}") logger.error(f"Fehler beim Verarbeiten von {url}: {e}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()