11f1444599
- Stufe 1 (/api/rohdaten): deterministischer Fetch + Persistenz pro PLZ/Woche in data/roh/, ohne LLM/Key. speicher.py serialisiert belegte Angebote verlustfrei (fehlende Felder bleiben null). - OpenRouter-Konfig als separates Panel (gilt für Stufe 2). - Stufe 2 (/api/kategorisieren): LLM-Schritt auf den GESPEICHERTEN Rohdaten, gesperrt solange keine vorliegen (400). Fetcht nicht erneut. - Funktionales Premium-Redesign: zwei nummerierte Stufen-Karten mit Status- Flags, erzwungene Reihenfolge, belegte Rohliste, ehrlicher Footer. - 47 Tests (+11: speicher Round-Trip, Endpoint-Sperre, Rohdaten offline). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
"""Persistenz der Rohdaten -- offline, deterministisch (kein Netz, kein LLM).
|
|
|
|
Geprüft wird der Architektur-Vertrag, nicht nur der Happy Path:
|
|
* Round-Trip ist verlustfrei für belegte Felder (inkl. fehlender Felder).
|
|
* Fehlende Felder bleiben None -- werden NICHT aufgefüllt.
|
|
* Kein Stand -> None (Stufe 2 ist dann gesperrt), es wird nichts geraten.
|
|
* Pfad liegt pro PLZ/Kalenderwoche.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
|
|
from angebote.modell import FetchErgebnis
|
|
from angebote.speicher import (
|
|
lade_rohdaten,
|
|
meta_fuer,
|
|
pfad_fuer,
|
|
rohliste_dicts,
|
|
speichere_rohdaten,
|
|
)
|
|
from tests.fakes import beispiel_angebot
|
|
|
|
|
|
def _fetch(angebote, plz="60487"):
|
|
return FetchErgebnis(
|
|
ort_plz=plz,
|
|
ort_name=None,
|
|
angebote=tuple(angebote),
|
|
abgedeckte_quellen=("marktguru",),
|
|
gesehene_haendler=tuple(sorted({a.haendler for a in angebote})),
|
|
hinweise=("marktguru: Teilabdeckung",),
|
|
)
|
|
|
|
|
|
def test_round_trip_behaelt_belegte_felder(tmp_path):
|
|
a = beispiel_angebot(
|
|
"Bio-Banane",
|
|
haendler="ALDI SÜD",
|
|
preis=1.29,
|
|
marke="Bio Smiley",
|
|
menge="1 kg",
|
|
gueltig_von=date(2026, 6, 1),
|
|
gueltig_bis=date(2026, 6, 7),
|
|
)
|
|
fetch = _fetch([a])
|
|
pfad = speichere_rohdaten(fetch, basis_dir=tmp_path)
|
|
assert pfad.exists()
|
|
|
|
geladen = lade_rohdaten("60487", basis_dir=tmp_path)
|
|
assert geladen is not None
|
|
assert len(geladen.angebote) == 1
|
|
g = geladen.angebote[0]
|
|
assert g.titel == "Bio-Banane"
|
|
assert g.haendler == "ALDI SÜD"
|
|
assert g.preis == 1.29
|
|
assert g.marke == "Bio Smiley"
|
|
assert g.gueltig_von == date(2026, 6, 1)
|
|
assert g.gueltig_bis == date(2026, 6, 7)
|
|
# stabile ID bleibt über den Round-Trip identisch (für Stufe-2-Mapping):
|
|
assert g.angebot_id == a.angebot_id
|
|
# Herkunft bleibt belegt:
|
|
assert geladen.abgedeckte_quellen == ("marktguru",)
|
|
assert geladen.gesehene_haendler == ("ALDI SÜD",)
|
|
|
|
|
|
def test_fehlende_felder_bleiben_none_kein_auffuellen(tmp_path):
|
|
# Pflichtfelder gesetzt, optionale bewusst leer -> dürfen NICHT geraten werden.
|
|
a = beispiel_angebot(
|
|
"No-Name-Artikel",
|
|
haendler="REWE",
|
|
preis=None,
|
|
marke=None,
|
|
menge=None,
|
|
grundpreis=None,
|
|
gueltig_von=None,
|
|
gueltig_bis=None,
|
|
)
|
|
speichere_rohdaten(_fetch([a]), basis_dir=tmp_path)
|
|
g = lade_rohdaten("60487", basis_dir=tmp_path).angebote[0]
|
|
assert g.preis is None
|
|
assert g.marke is None
|
|
assert g.menge is None
|
|
assert g.grundpreis is None
|
|
assert g.gueltig_von is None
|
|
assert g.gueltig_bis is None
|
|
|
|
|
|
def test_kein_stand_liefert_none(tmp_path):
|
|
assert lade_rohdaten("12345", basis_dir=tmp_path) is None
|
|
assert meta_fuer("12345", basis_dir=tmp_path) is None
|
|
|
|
|
|
def test_meta_fasst_zusammen(tmp_path):
|
|
a = beispiel_angebot("Butter", haendler="REWE")
|
|
b = beispiel_angebot("Käse", haendler="EDEKA")
|
|
speichere_rohdaten(_fetch([a, b]), basis_dir=tmp_path)
|
|
meta = meta_fuer("60487", basis_dir=tmp_path)
|
|
assert meta["anzahl"] == 2
|
|
assert set(meta["haendler"]) == {"REWE", "EDEKA"}
|
|
assert meta["abgerufen_am"]
|
|
|
|
|
|
def test_pfad_pro_plz_und_woche():
|
|
jetzt = datetime(2026, 6, 2, 9, 0, 0) # ISO-Woche 23/2026
|
|
pfad = pfad_fuer("60487", basis_dir="/tmp/roh", jetzt=jetzt)
|
|
assert pfad.name == "60487_2026-W23.json"
|
|
|
|
|
|
def test_rohliste_ohne_produktgruppe(tmp_path):
|
|
"""Stufe 1 kategorisiert nicht -- die Rohliste trägt keine Produktgruppe."""
|
|
a = beispiel_angebot("Butter")
|
|
dicts = rohliste_dicts(_fetch([a]))
|
|
assert dicts and "produktgruppe" not in dicts[0]
|
|
assert "gruppe" not in dicts[0]
|