aktualisiert nach erstem Fehler wegen veralteter sqlalchemy
This commit is contained in:
parent
18b177a58d
commit
6a4b3b6681
|
|
@ -5,9 +5,9 @@ import hashlib
|
|||
from datetime import datetime
|
||||
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.ext.declarative import declarative_base
|
||||
from sqlalchemy.orm import sessionmaker, relationship
|
||||
|
||||
# Basis-Klasse für SQLAlchemy-Modelle
|
||||
Base = declarative_base()
|
||||
|
|
@ -49,29 +49,25 @@ class BundleSalesHistory(Base):
|
|||
bundle = relationship("Bundle", back_populates="sales_history")
|
||||
|
||||
def calculate_hash(data: dict) -> str:
|
||||
"""
|
||||
Berechnet einen SHA-256 Hash aus dem 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)
|
||||
return hashlib.sha256(json_string.encode('utf-8')).hexdigest()
|
||||
|
||||
def fetch_bundle_data(url: str) -> dict:
|
||||
"""
|
||||
Ruft die Bundle-Detailseite ab und extrahiert den JSON-Inhalt aus dem <script>-Tag mit der ID "webpack-bundle-page-data".
|
||||
"""
|
||||
"""Lädt die Detailseite eines Bundles und extrahiert den JSON-Inhalt aus dem <script>-Tag 'webpack-bundle-page-data'."""
|
||||
response = requests.get(url)
|
||||
response.raise_for_status()
|
||||
soup = BeautifulSoup(response.text, "html.parser")
|
||||
script_tag = soup.find("script", {"id": "webpack-bundle-page-data", "type": "application/json"})
|
||||
if not script_tag:
|
||||
raise ValueError("Kein JSON-Datenblock gefunden auf der Detailseite!")
|
||||
raise ValueError("Kein JSON-Datenblock 'webpack-bundle-page-data' gefunden auf der Detailseite!")
|
||||
return json.loads(script_tag.string)
|
||||
|
||||
def process_bundle(session, url: str):
|
||||
"""
|
||||
Verarbeitet ein einzelnes Bundle:
|
||||
- Lädt die Detailseite und extrahiert den JSON-Datensatz
|
||||
- Berechnet einen Hash des relevanten Datenbereichs (hier: bundleData)
|
||||
- Lädt die Detailseite und extrahiert den JSON-Datensatz (bundleData)
|
||||
- Berechnet einen Hash des relevanten Datenbereichs
|
||||
- Vergleicht mit der letzten Version in der DB und speichert bei Änderung eine neue Version
|
||||
- Speichert Verkaufszahlen in einer separaten Tabelle
|
||||
"""
|
||||
|
|
@ -85,7 +81,7 @@ def process_bundle(session, url: str):
|
|||
machine_name = bundle_data.get("machine_name", "")
|
||||
human_name = bundle_data.get("human_name", "")
|
||||
|
||||
# Definiere den relevanten Datenausschnitt; hier nehmen wir das gesamte bundleData
|
||||
# Relevanter Datenausschnitt – hier nehmen wir das gesamte bundleData
|
||||
relevant_data = bundle_data
|
||||
new_hash = calculate_hash(relevant_data)
|
||||
|
||||
|
|
@ -96,13 +92,12 @@ def process_bundle(session, url: str):
|
|||
session.add(bundle)
|
||||
session.commit()
|
||||
|
||||
# Hole die aktuellste Version (nach timestamp sortiert)
|
||||
# Hole die aktuellste Version dieses Bundles (nach timestamp sortiert)
|
||||
latest_version = (session.query(BundleVersion)
|
||||
.filter_by(bundle_id=bundle.id)
|
||||
.order_by(BundleVersion.timestamp.desc())
|
||||
.first())
|
||||
|
||||
# Wenn es noch keine Version gibt oder der Hash sich geändert hat, speichere eine neue Version
|
||||
if latest_version is None or latest_version.version_hash != new_hash:
|
||||
new_version = BundleVersion(
|
||||
bundle_id=bundle.id,
|
||||
|
|
@ -117,10 +112,11 @@ def process_bundle(session, url: str):
|
|||
else:
|
||||
print(f"Bundle '{human_name}' hat sich nicht geändert.")
|
||||
|
||||
# Verkaufszahlen extrahieren – Annahme: Verkaufszahlen stehen unter "bundles_sold|decimal" entweder direkt in bundleData oder in basic_data
|
||||
bundles_sold = bundle_data.get("bundles_sold|decimal", None)
|
||||
# Verkaufszahlen extrahieren – hier wird angenommen, dass sie entweder direkt in bundleData
|
||||
# oder unter basic_data stehen, im Feld "bundles_sold|decimal"
|
||||
bundles_sold = bundle_data.get("bundles_sold|decimal")
|
||||
if bundles_sold is None:
|
||||
bundles_sold = bundle_data.get("basic_data", {}).get("bundles_sold|decimal", None)
|
||||
bundles_sold = bundle_data.get("basic_data", {}).get("bundles_sold|decimal")
|
||||
if bundles_sold is not None:
|
||||
try:
|
||||
sales_value = float(bundles_sold)
|
||||
|
|
@ -136,20 +132,29 @@ def process_bundle(session, url: str):
|
|||
|
||||
def get_bundle_urls(overview_url: str) -> list:
|
||||
"""
|
||||
Ruft die Übersichtsseite ab und extrahiert alle Bundle-URLs.
|
||||
Hier wird angenommen, dass Bundle-Links in <a>-Tags mit der CSS-Klasse "bundle-link" stehen.
|
||||
Passe den Selektor ggf. an die tatsächliche Seitenstruktur an.
|
||||
Ruft die Übersichtsseite ab und extrahiert alle Bundle-URLs aus dem JSON-Datenblock
|
||||
im <script>-Tag mit der ID "landingPage-json-data". Dies ist zuverlässig, da die
|
||||
Bundles dort dynamisch eingebunden werden.
|
||||
"""
|
||||
response = requests.get(overview_url)
|
||||
response.raise_for_status()
|
||||
soup = BeautifulSoup(response.text, "html.parser")
|
||||
bundle_urls = []
|
||||
# Beispiel: Alle <a>-Tags mit class="bundle-link"
|
||||
for a in soup.find_all("a", class_="bundle-link"):
|
||||
href = a.get("href")
|
||||
if href:
|
||||
full_url = requests.compat.urljoin(overview_url, href)
|
||||
landing_script = soup.find("script", {"id": "landingPage-json-data", "type": "application/json"})
|
||||
if landing_script:
|
||||
landing_data = json.loads(landing_script.string)
|
||||
# Wir erwarten, dass in landing_data["data"] die Kategorien "books", "games", "software" enthalten sind.
|
||||
for category in ["books", "games", "software"]:
|
||||
cat_data = landing_data.get("data", {}).get(category, {})
|
||||
for section in cat_data.get("mosaic", []):
|
||||
for product in section.get("products", []):
|
||||
# Annahme: Der Link steht im Feld "product_url"
|
||||
url = product.get("product_url", "")
|
||||
if url:
|
||||
full_url = requests.compat.urljoin(overview_url, url)
|
||||
bundle_urls.append(full_url)
|
||||
else:
|
||||
print("Kein JSON-Datenblock 'landingPage-json-data' auf der Übersichtsseite gefunden.")
|
||||
return bundle_urls
|
||||
|
||||
def main():
|
||||
|
|
@ -159,19 +164,19 @@ def main():
|
|||
Session = sessionmaker(bind=engine)
|
||||
session = Session()
|
||||
|
||||
# URL der Übersichtsseite, von der alle Bundles erfasst werden sollen
|
||||
# URL der Bundles-Übersichtsseite
|
||||
overview_url = "https://www.humblebundle.com/bundles"
|
||||
|
||||
try:
|
||||
bundle_urls = get_bundle_urls(overview_url)
|
||||
if not bundle_urls:
|
||||
print("Keine Bundle-URLs gefunden! Überprüfe den CSS-Selektor in get_bundle_urls().")
|
||||
print("Keine Bundle-URLs gefunden! Überprüfe den JSON-Datenblock oder den Selektor in get_bundle_urls().")
|
||||
return
|
||||
except Exception as e:
|
||||
print(f"Fehler beim Abrufen der Übersichtseite: {e}")
|
||||
print(f"Fehler beim Abrufen der Übersichtsseite: {e}")
|
||||
return
|
||||
|
||||
# Iteriere über alle gefundenen Bundle-URLs und verarbeite jedes Bundle
|
||||
# Verarbeite alle gefundenen Bundle-URLs
|
||||
for url in bundle_urls:
|
||||
print(f"Verarbeite Bundle: {url}")
|
||||
try:
|
||||
|
|
|
|||
Loading…
Reference in New Issue