Initiale Implementierung: Angebots-Übersicht (Fetch + Kategorisierung + Web-UI)

Der Schnitt aus CLAUDE.md ist durchgehalten:
- Fetch (deterministisch, kein LLM): marktguru-Adapter mit geprüftem
  Ortsbezug (zipCode), Wochen-Cache, robots.txt-Respekt, ehrlicher Regel-4-
  Abbruch bei fehlendem Beleg statt Krücke.
- Kategorisierung (einziger LLM-Ort): geschlossene Liste + Daten-Integrität
  als Code erzwungen; austauschbar via Protokoll (OpenRouter/Anthropic),
  mit Drosselung/Retry und ehrlichem Abbruch.
- FastAPI-Web-UI als dünne Schicht: Modellauswahl (Liste/Suche/Refresh),
  Live-Fortschritt, gruppierte Ergebnisse mit Filtern, Ergebnis-Cache.
- 36 Tests gegen die Architektur-Regeln (kein Auffüllen, Abbruch, Integrität,
  geschlossene Liste, Unsicherheit, Schnitt) und die Web-Schicht.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jeuner 2026-06-03 09:29:59 +02:00
commit 39b8a98bc2
34 changed files with 3215 additions and 0 deletions

View file

@ -0,0 +1,125 @@
---
name: angebote-fetch
description: >
Deterministischer Abruf wöchentlicher Supermarkt-/Discounter-Angebote für
einen konkreten Ort. Verwenden, wenn aktuelle Angebotsdaten beschafft werden
sollen. Dieser Skill enthält KEINE LLM-Aufrufe -- er holt, parst und
normalisiert nur. Trigger: "Angebote holen", "Daten für Ort X", "Prospekte
abrufen", Aufbau/Pflege der Datenbeschaffung.
---
# angebote-fetch
## Zweck
Beschaffe für einen gegebenen Ort (PLZ oder Ortsname) die aktuell gültigen
Angebote, händlerübergreifend, und gib sie als normalisierten, belegten
Datensatz zurück. Nichts wird interpretiert, kategorisiert oder bewertet --
das ist Aufgabe des Skills `angebote-kategorisieren`.
## Harte Regeln (Architektur, nicht Präferenz)
Diese Regeln gelten unabhängig von Kontext und Plausibilität:
1. **Keine LLM-Aufrufe.** Dieser Teil ist rein deterministisch. Wenn du
versucht bist, ein Modell "zum Aufräumen" einzusetzen, ist das der falsche
Skill -- halte an.
2. **Nur Belegtes.** Jedes zurückgegebene Angebot trägt seine Quelle (Händler,
Quell-URL/-ID, Abrufzeitpunkt). Felder, die in der Quelle fehlen, werden als
`null`/fehlend markiert -- niemals geraten oder geschätzt.
3. **Kein Auffüllen.** Wenn eine Quelle nichts liefert, wird das als leeres
Ergebnis dieser Quelle gemeldet, nicht durch Beispiele ersetzt.
4. **Abbruch bei Spezifitätsmangel.** Wenn der Ort nicht aufgelöst werden kann
oder keine Quelle für den Ort filtert, brich mit einer klaren Meldung ab:
Schwelle (welche), Ursache (Ort nicht auflösbar / keine Quelle deckt Ort ab),
konkreter Vorschlag (z. B. größerer Ort in der Nähe, andere Quelle).
Liefere kein ortsfremdes Ergebnis als Notlösung.
## Datenmodell
Ein normalisiertes Angebot (Vorschlag, anpassbar):
```python
@dataclass
class Angebot:
titel: str # Produktname, wie in der Quelle
marke: str | None # falls vorhanden
preis: float | None # in EUR; None wenn nicht eindeutig parsebar
grundpreis: str | None # z. B. "1 kg = 4,44 EUR", roh übernommen
menge: str | None # z. B. "200g Packung", roh übernommen
gueltig_von: date | None
gueltig_bis: date | None
haendler: str # Pflicht
quelle: str # URL oder Quell-ID, Pflicht
abgerufen_am: datetime # Pflicht
# bewusst KEIN feld "produktgruppe" -- das setzt der andere skill
```
Das Fehlen eines `produktgruppe`-Felds ist Absicht: Die Trennung der beiden
Verantwortungen ist im Datenmodell verankert. In der Implementierung ist
`Angebot` zusätzlich **eingefroren** (`frozen=True`) -- der Kategorisier-Schritt
*kann* die Daten damit nicht verändern, nicht nur *soll* es nicht.
## Quellen
Kandidaten sind Angebots-Aggregatoren mit Ortsfilter. Wichtige bekannte
Eigenheiten:
- Der Ortsfilter steckt bei vielen Aggregatoren in Session/Cookie, nicht in der
URL -- ein roher Abruf ohne Standort liefert ggf. bundesweite Angebote.
Stelle den Ortsbezug explizit her und prüfe ihn, statt ihm zu vertrauen.
- Manche Quellen liefern strukturierte Felder (Titel, Marke, Preis, Gültigkeit,
Händler) gut parsebar; andere nur als Bild-Prospekt. Bild-Prospekte sind
außerhalb des Scopes dieses Skills, solange keine strukturierte Quelle
existiert -- in dem Fall greift Regel 4 (Abbruch + ehrlicher Hinweis).
- Discounter (Aldi, Lidl) sind bei Aggregatoren oft unterrepräsentiert. Das
ist ein Abdeckungsloch, das im Ergebnis ehrlich ausgewiesen wird
("Discounter X nicht abgedeckt"), nicht kaschiert.
Halte die konkrete Quellenliste in einem Config-/Adapter-Modul, damit Quellen
ausgetauscht werden können, ohne die Kernlogik anzufassen. Ein Adapter pro
Quelle (gleiche Schnittstelle: rein = Ort, raus = Liste[Angebot]).
## Erwartetes Scheitern (Regel 4 ist der Normalfall, nicht der Ausnahmefall)
Beim Abruf realer Aggregatoren ist Scheitern eingeplant, nicht überraschend.
Die folgenden Fälle sind **vorgesehene Abbruchfälle** -- es wird kein brüchiger
Workaround gebaut, der "irgendwie etwas" liefert:
- **Ortsfilter in Session/Cookie statt URL.** Liefert eine Quelle trotz
gesetztem Ort erkennbar bundesweite statt ortsbezogene Angebote, ist der
Ortsbezug *nicht verifiziert*. Dann bricht der Adapter ab (Regel 4) --
er gibt nicht ein bundesweites Ergebnis als "Ort X" aus.
- **Nur Bild-Prospekte.** Liefert eine Quelle ausschließlich Prospekt-Bilder
ohne strukturierte Felder, ist sie außerhalb des Scopes. Kein OCR-Raten,
keine geschätzten Preise -- die Quelle meldet "keine strukturierten Daten"
und wird übersprungen oder führt (wenn sie die einzige Quelle war) zum
Abbruch mit Hinweis.
- **Zugang/Schlüssel fehlt oder bricht weg.** Lässt sich der für eine Quelle
nötige Zugang (z. B. ein aus der Seite gelesener Client-/API-Schlüssel) nicht
zuverlässig herstellen, ist das ein Abbruchgrund -- kein hartkodierter,
geratener Schlüssel.
Merksatz: **Den ehrlichen Abbruch bauen, nicht die Krücke.** Ein Abbruch mit
klarer Ursache und Vorschlag ist ein korrektes Ergebnis dieses Skills. Ein
plausibel aussehender, aber unbelegt zusammengeflickter Datensatz ist ein
Fehler -- auch wenn er "funktioniert".
Die Verifikation des Ortsbezugs ist deshalb selbst eine prüfbare Bedingung im
Datenfluss: Der Adapter belegt, *dass* und *womit* er den Ort gefiltert hat
(z. B. zurückgegebener `zipCode` im Request, Händler-/Filiale-Bezug in der
Antwort), und bricht ab, wenn dieser Beleg fehlt.
## Robustheit
- Respektiere robots.txt und vernünftige Request-Raten; cache Ergebnisse pro
Ort/Woche, statt bei jedem Lauf neu zu ziehen.
- Schreibe Tests für: leeres Quellergebnis (→ kein Auffüllen), nicht
auflösbarer Ort (→ Abbruch mit Meldung), fehlende Felder (→ als fehlend
markiert, nicht geraten).
## Was dieser Skill NICHT tut
- Keine Produktgruppen-Zuordnung.
- Keine Bewertung "guter"/"schlechter" Angebote.
- Keine sprachliche Aufbereitung. Nur Beschaffung und Normalisierung.

View file

@ -0,0 +1,102 @@
---
name: angebote-kategorisieren
description: >
Ordnet einen flachen, bereits beschafften Angebots-Stream in saubere
Produktgruppen. Dies ist der EINZIGE Teil des Projekts, in dem ein LLM
eingesetzt wird -- weil die Einordnung genuin unscharf ist. Verwenden, wenn
normalisierte Angebote (aus angebote-fetch) in eine nach Kategorien
geordnete Übersicht gebracht werden sollen. Trigger: "kategorisieren",
"nach Produktgruppen sortieren", "Übersicht bauen".
---
# angebote-kategorisieren
## Zweck
Nimm die normalisierten, belegten Angebote aus `angebote-fetch` und ordne jedes
einer Produktgruppe zu. Ergebnis ist eine nach Kategorien gruppierte Übersicht.
Dieser Skill **verändert keine Angebotsdaten** -- er fügt nur die
Gruppenzuordnung hinzu.
## Warum hier ein LLM sinnvoll ist (und nur hier)
Die Zuordnung ist die echte Ambiguität der Aufgabe. Regelbasiert ("enthält
'Käse' → Molkereiprodukte") scheitert an Toffifee, Schwarzwälder Schinken,
Grundnahrungsmitteln mit Markennamen, Non-Food in Supermarktprospekten. Diese
unscharfe Einordnung ist exakt die Stärke eines Sprachmodells. Deshalb läuft
**nur** dieser Schritt über ein LLM -- nicht die Datenbeschaffung.
## Harte Regeln (Architektur, nicht Präferenz)
1. **Daten sind unantastbar.** Das LLM darf Titel, Preis, Gültigkeit, Händler
NICHT verändern, korrigieren oder ergänzen. Es vergibt ausschließlich eine
Produktgruppe. Wenn ein Preis fehlt, bleibt er fehlend -- nicht "ergänzen".
2. **Geschlossene Kategorienliste.** Das Modell wählt aus einer fest
definierten Liste von Produktgruppen (s. u.). Es erfindet keine neuen
Kategorien. Passt ein Artikel in keine, kommt er in `Sonstiges` bzw.
`Non-Food`.
3. **Eine Gate-Prüfung pro Angebot.** Jedes Angebot durchläuft die Zuordnung
einzeln. Wenn das Modell bei einem Artikel unsicher ist, markiert es ihn als
`unsicher: true` mit der wahrscheinlichsten Gruppe -- es rät nicht still.
4. **Kein stilles Absenken bei vielen Elementen.** Bei langen Listen darf die
Zuordnungsqualität nicht zum Ende hin abfallen. Verarbeite in kleinen
Batches mit gleichbleibendem Schema; fülle nie eine Gruppe auf, um sie
"ausgewogen" aussehen zu lassen.
## Produktgruppen (Startliste, anpassbar)
```
Obst & Gemüse
Fleisch & Wurst
Fisch
Molkereiprodukte & Eier
Brot & Backwaren
Grundnahrungsmittel (Nudeln, Reis, Mehl, Konserven)
Süßwaren & Snacks
Getränke (alkoholfrei)
Alkoholische Getränke
Tiefkühl
Drogerie & Haushalt
Non-Food (Kleidung, Spielzeug, Garten, Technik)
Sonstiges
```
Die Liste lebt in einer Config, nicht im Prompt-Text -- so kann das Team sie
anpassen, ohne die Logik zu berühren.
## LLM-Aufruf
- Übergib dem Modell pro Batch nur das Nötige: Titel, Marke, Menge. NICHT
Preis/Gültigkeit (die sind für die Einordnung irrelevant und sollen nicht
versehentlich verändert zurückkommen).
- Lass das Modell strukturiert antworten (z. B. JSON: `{id, gruppe,
unsicher}`), und mappe das Ergebnis zurück auf die unveränderten
Original-Angebote über eine stabile ID. Übernimm vom Modell **nur** Gruppe
und Unsicherheits-Flag -- alles andere kommt aus dem Originaldatensatz.
- System-Vorgabe an das Modell: geschlossene Kategorienliste, keine neuen
Kategorien, bei Unsicherheit Flag setzen statt raten.
- Liefert das Modell trotzdem eine Gruppe außerhalb der Liste, wird sie nicht
übernommen: der Artikel landet in `Sonstiges` und wird als `unsicher`
markiert. Die geschlossene Liste ist eine prüfbare Bedingung im Code, keine
Bitte an das Modell.
## Ausgabe
Eine nach Produktgruppen geordnete Übersicht. Pro Gruppe die zugehörigen
Angebote mit ihren belegten Feldern. Leere Gruppen werden als "keine Angebote"
ausgewiesen -- nicht weggelassen und nicht aufgefüllt. Unsicher zugeordnete
Artikel werden sichtbar markiert, damit ein Mensch nachsehen kann.
## Tests
- Daten-Integrität: Preis/Gültigkeit/Händler nach Kategorisierung identisch zu
vorher (Property-Test über den ganzen Stream).
- Geschlossene Liste: keine Gruppe außerhalb der Config taucht auf.
- Unsicherheit: bewusst mehrdeutige Artikel (z. B. "Pflanzendrink") werden
geflaggt, nicht still einsortiert.
## Was dieser Skill NICHT tut
- Keine Datenbeschaffung (das macht angebote-fetch).
- Keine Preis-/Qualitätsbewertung.
- Kein Verändern, Reparieren oder Ergänzen von Angebotsdaten.

14
.gitignore vendored Normal file
View file

@ -0,0 +1,14 @@
.venv/
__pycache__/
*.pyc
.cache/
.pytest_cache/
.DS_Store
# generierte Ausgaben / lokale Artefakte
*.err
/angebote-*.md
# Original-Input-Archive (Specs liegen kanonisch in .claude/skills/)
/files.zip
/temp/

90
CLAUDE.md Normal file
View file

@ -0,0 +1,90 @@
# CLAUDE.md -- Angebots-Übersicht
Leitfaden für die Entwicklung dieses Projekts mit Claude Code.
## Was dieses Projekt ist
Eine sachliche, ortskonkrete Übersicht der wöchentlichen Supermarkt- und
Discounter-Angebote, sortiert nach Produktgruppen. Kein Hochglanzprospekt
einzelner Händler, sondern eine händlerübergreifende, neutrale Liste:
"Was ist diese Woche an Ort X im Angebot, geordnet nach Kategorie."
## Das Leitprinzip: der Schnitt
Diese Aufgabe sieht nach einer KI-Aufgabe aus. Sie ist es nur zur Hälfte.
Sie zerfällt in zwei grundverschiedene Teilprobleme, die strikt getrennt
gehören:
1. **Daten holen** (ortskonkret, händlerübergreifend) -- das ist
**deterministisch**. Hier wird KI eingespart. Ein HTTP-Abruf / Parser
leistet das exakt und reproduzierbar. Ein LLM würde hier nur Preise und
Gültigkeiten halluzinieren. Dieser Teil enthält **keine** LLM-Aufrufe.
2. **Kategorisieren** (flacher Angebots-Stream → saubere Produktgruppen) --
das ist die **echte Ambiguität**: "Ist Toffifee Süßwaren? Ist Schwarzwälder
Schinken Fleisch/Wurst? Gehört eine Blühpflanze überhaupt in eine
Lebensmittel-Übersicht?" Genau hier ist ein LLM das richtige Werkzeug --
und nur hier.
> Merksatz: Vor KI muss man erst einmal KI einsparen. Aber dort, wo nur sie
> passt, muss man sie auch einsetzen. Wer die ganze Aufgabe dem Modell gibt,
> bekommt Fiktion. Wer sie ganz deterministisch löst, scheitert an der
> Kategorisierung. Richtig ist die Arbeitsteilung.
Dieser Schnitt ist nicht verhandelbar. Er ist die Architektur des Projekts,
nicht eine Stilpräferenz. Vermische die beiden Teile nicht.
## Architektur statt Präferenz
Qualitätsregeln werden hier **festverdrahtet**, nicht "mitgedacht". Eine Regel,
die nur als höfliche Bitte im Code steht ("möglichst keine erfundenen Preise"),
bricht unter Druck still weg. Eine Regel, die als prüfbare Bedingung im
Datenfluss steht, hält.
Konkrete Konsequenzen, die jederzeit gelten:
- **Kein Auffüllen.** Wenn für eine Produktgruppe keine belegten Angebote
vorliegen, wird "keine Daten" ausgegeben -- niemals ein plausibel klingendes
Beispiel erfunden.
- **Jedes ausgegebene Angebot ist belegt.** Preis, Gültigkeit und Händler
stammen aus dem abgerufenen Datensatz, nicht aus dem Modell. Wenn ein Feld
fehlt, wird es als fehlend markiert, nicht geraten.
- **Abbruch statt stiller Drift.** Wenn die Datenlage die Anforderung nicht
hergibt (z. B. kein Treffer für den Ort), bricht das Programm mit einer
klaren Meldung ab und nennt die Ursache und einen Erweiterungsvorschlag.
Es liefert kein "irgendwie vollständig aussehendes" Ergebnis.
Diese drei Punkte sind der Grund, warum "die KI soll sagen, wenn sie scheitert"
hier funktioniert: nicht weil das Modell Einsicht hätte, sondern weil eine
externe, prüfbare Bedingung es erzwingt.
## Skills
Das Projekt nutzt zwei Skills (in `.claude/skills/`), die exakt dem Schnitt
folgen:
- **angebote-fetch** -- deterministischer Datenabruf. Keine LLM-Aufrufe.
- **angebote-kategorisieren** -- LLM-gestützte Einordnung in Produktgruppen.
Lies die jeweilige `SKILL.md`, bevor du an dem zugehörigen Teil arbeitest.
Die SKILL.md ist die verbindliche Spezifikation für ihren Bereich.
## Tech-Kontext
- Sprache: **Python** (FastAPI-nah).
- Öffentlicher Stack-Bezug, falls relevant: FastAPI, Qdrant, A2A.
- Datenquellen-Kandidaten für den Fetch-Teil: Angebots-Aggregatoren, die nach
Ort/PLZ filtern (z. B. marktguru, kaufda, MeinProspekt). Discounter wie
Aldi/Lidl liefern oft nicht an Aggregatoren -- das ist ein bekanntes
Abdeckungsloch und gehört ehrlich als solches ausgewiesen, nicht kaschiert.
## Arbeitsweise mit Claude Code
- Beginne jede Aufgabe damit, den relevanten Teil dieses Dokuments und die
passende SKILL.md zu lesen.
- Halte den Schnitt sauber: kein LLM-Aufruf im Fetch-Teil, kein heimliches
Daten-"Reparieren" im Kategorisier-Teil.
- Wenn eine Anforderung den Schnitt verletzen würde, benenne den Konflikt,
bevor du Code schreibst.
- Schreibe Tests für die Architektur-Regeln (kein Auffüllen, Abbruch bei
leerer Datenlage), nicht nur für den Happy Path.

102
README.md Normal file
View file

@ -0,0 +1,102 @@
# Angebots-Übersicht
Ortskonkrete, händlerübergreifende Übersicht wöchentlicher Supermarkt-Angebote,
geordnet nach Produktgruppen. Neutral, nicht als Hochglanzprospekt.
## Der Gedanke
Die Aufgabe sieht nach einer KI-Aufgabe aus, ist es aber nur zur Hälfte. Sie
zerfällt in zwei strikt getrennte Teile:
1. **Daten holen** -- deterministisch, ohne LLM. (Skill `angebote-fetch`)
2. **Kategorisieren** -- die echte Ambiguität, hier gehört das LLM hin.
(Skill `angebote-kategorisieren`)
Wer alles dem Modell gibt, bekommt erfundene Preise. Wer alles deterministisch
löst, scheitert an der Einordnung. Die Architektur erzwingt den Schnitt.
Der zweite Punkt, der dieses Projekt trägt: Qualitätsregeln stehen nicht als
gut gemeinte Bitten im Code, sondern als prüfbare Bedingungen -- kein
Auffüllen, nur Belegtes, Abbruch statt stiller Drift. Genau dadurch "sagt das
System, wenn es scheitert": nicht aus Einsicht des Modells, sondern weil eine
externe Bedingung es erzwingt.
## Entwicklung mit Claude Code
`CLAUDE.md` ist der verbindliche Leitfaden. Die beiden `SKILL.md` in
`.claude/skills/` sind die Spezifikationen der zwei Teile. Claude Code liest
sie automatisch -- beim Arbeiten am jeweiligen Teil zuerst die passende
SKILL.md lesen.
## Setup
```bash
python3 -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
```
## Nutzung
```bash
# PLZ direkt (immer verlässlich):
python -m angebote 60487
# Ortsname (nur für die in config.py hinterlegte Auswahl großer Städte;
# unbekannter Ort -> ehrlicher Abbruch mit Vorschlag, keine Notlösung):
python -m angebote "Frankfurt"
# Ohne Kategorisierung (kein LLM, flache belegte Liste):
python -m angebote 60487 --no-llm
```
Der Kategorisier-Schritt braucht `ANTHROPIC_API_KEY` in der Umgebung. Fehlt er
und es wird kein `--no-llm` gesetzt, bricht das Programm ehrlich ab, statt
ungeordnet weiterzulaufen.
## Stand der Implementierung (ehrlich)
- `requirements.txt` -- **vorhanden**.
- `src/angebote/` -- **vorhanden**: Datenmodell, Adapter-Schnittstelle,
Fetch-Orchestrator, Kategorisier-Schritt, Übersicht-Renderer, CLI.
- `tests/` -- **vorhanden**: Architektur-Tests (kein Auffüllen, Abbruch bei
leerem/unauflösbarem Ort, Daten-Integrität nach Kategorisierung,
geschlossene Kategorienliste, Unsicherheits-Flag, Schnitt-Test "kein LLM im
Fetch-Teil"). Laufen offline, ohne Netz und ohne LLM.
- **Live bestätigt:** der marktguru-Adapter wurde gegen die echte API getestet
(PLZ 60487) und liefert reale, belegte Angebote (u. a. ALDI SÜD, PENNY, Lidl,
REWE, Kaufland, nahkauf). Erkenntnisse aus dem echten Lauf, die direkt in den
Code geflossen sind:
- `offers/search` ist query-orientiert -- leeres `q` liefert 0 Treffer. Der
Adapter aggregiert daher über Kategorie-Seedbegriffe (config) und weist
diese **Teilabdeckung** ehrlich aus, statt Vollständigkeit zu behaupten.
- Die pauschale Annahme "Aldi/Lidl fehlen bei Aggregatoren" wurde von den
Daten **widerlegt** (beide sind enthalten). Die Abdeckung wird deshalb
**datengetrieben** ausgewiesen (beobachtete Händler), nicht hartkodiert.
- marktgurus Marken-Sentinel `thisisnobrand123` wird als "keine Marke"
behandelt, nicht als Beleg durchgereicht.
- **Voraussetzungen für den Live-Lauf:** installiertes `requests` (certifi für
TLS), erreichbares Netz, von der Seite lesbare API-Schlüssel. Fehlt eines
davon, ist das der vorgesehene Abbruchfall (Regel 4) mit *zutreffend*
benannter Ursache -- keine Krücke.
## Struktur
```
CLAUDE.md Leitfaden / Architektur
README.md dieses Dokument
requirements.txt Abhängigkeiten
.claude/skills/angebote-fetch/ Spec: deterministischer Datenabruf
.claude/skills/angebote-kategorisieren/ Spec: LLM-gestützte Einordnung
src/angebote/ Implementierung
modell.py eingefrorenes Angebot-Datenmodell
fehler.py AbbruchFehler (Regel 4)
config.py Quellenliste, Produktgruppen, Orts-Auflösung
quellen/ ein Adapter pro Quelle (rein = Ort, raus = [Angebot])
basis.py Adapter-Schnittstelle + Ort
marktguru.py erster echter Adapter, geprüfter Ortsbezug
fetch.py Orchestrator (Ort rein, belegte Angebote raus)
kategorisieren.py LLM-Schritt hinter einer Schnittstelle (offline testbar)
uebersicht.py Gruppierung + Rendering
cli.py / __main__.py CLI-Einstieg
tests/ Architektur-Tests
```

16
pyproject.toml Normal file
View file

@ -0,0 +1,16 @@
[project]
name = "angebote-uebersicht"
version = "0.1.0"
description = "Ortskonkrete, händlerübergreifende Angebots-Übersicht nach Produktgruppen"
requires-python = ">=3.10"
dependencies = [
"requests>=2.31",
"anthropic>=0.40",
]
[project.optional-dependencies]
dev = ["pytest>=8.0"]
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]

15
requirements.txt Normal file
View file

@ -0,0 +1,15 @@
# Fetch-Teil (deterministisch, KEIN LLM) UND OpenRouter-Kategorisierung:
requests>=2.31
# Kategorisier-Teil, nur für den NATIVEN Anthropic-Anbieter nötig.
# Bei Nutzung von OpenRouter (OPENROUTER_API_KEY) NICHT erforderlich -- der
# OpenRouter-Kategorisierer kommt mit requests aus.
anthropic>=0.40
# Web-UI (lokale FastAPI-App):
fastapi>=0.110
uvicorn>=0.29
# Entwicklung/Tests:
pytest>=8.0
httpx>=0.27 # für fastapi.testclient

10
src/angebote/__init__.py Normal file
View file

@ -0,0 +1,10 @@
"""Angebots-Übersicht: deterministischer Fetch + LLM-Kategorisierung.
Der Schnitt (siehe CLAUDE.md) ist auch in der Paketstruktur sichtbar:
* fetch / quellen / modell -> deterministisch, kein LLM
* kategorisieren -> der EINZIGE Ort mit LLM
"""
from .modell import Angebot, FetchErgebnis, KategorisiertesAngebot
__all__ = ["Angebot", "FetchErgebnis", "KategorisiertesAngebot"]

3
src/angebote/__main__.py Normal file
View file

@ -0,0 +1,3 @@
from .cli import main
raise SystemExit(main())

167
src/angebote/cli.py Normal file
View file

@ -0,0 +1,167 @@
"""CLI-Einstieg: Ort rein, geordnete Übersicht raus.
Der Schnitt bleibt auch hier sichtbar: erst `fetch` (deterministisch), dann --
nur falls gewünscht und Schlüssel vorhanden -- `kategorisieren` (LLM).
"""
from __future__ import annotations
import argparse
import os
import sys
from .fehler import AbbruchFehler
from .fetch import hole_angebote
from .modell import KategorisiertesAngebot
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
prog="angebote",
description="Ortskonkrete, händlerübergreifende Angebots-Übersicht nach Produktgruppen.",
)
parser.add_argument(
"ort",
nargs="?",
help="PLZ (5-stellig) oder hinterlegter Ortsname",
)
parser.add_argument(
"--no-llm",
action="store_true",
help="ohne Kategorisierung: flache, belegte Liste (kein LLM, kein Schlüssel nötig)",
)
parser.add_argument(
"--anbieter",
choices=("openrouter", "anthropic"),
default=None,
help="LLM-Anbieter erzwingen; ohne Angabe automatisch nach gesetztem Key",
)
parser.add_argument(
"--modell",
default=None,
help="Modell-ID überschreiben (Default je Anbieter)",
)
parser.add_argument(
"--modelle",
nargs="?",
const="",
default=None,
metavar="SUCHE",
help="OpenRouter-Modelle auflisten und beenden; optional Suchbegriff "
"(ohne Begriff: Top-5 Free, tool-fähig)",
)
parser.add_argument(
"--modell-waehlen",
action="store_true",
help="vor dem Lauf das OpenRouter-Modell interaktiv wählen (Liste/Suche/Update)",
)
args = parser.parse_args(argv)
# Reiner Listen-/Suchmodus -- braucht keinen Ort und keinen Key.
if args.modelle is not None:
return _liste_modelle(args.modelle)
if not args.ort:
parser.error("Ort fehlt (PLZ oder Ortsname) -- oder nutze --modelle zum Auflisten.")
try:
fetch = hole_angebote(args.ort)
except AbbruchFehler as e:
print(e.als_text(), file=sys.stderr)
return 2
if args.no_llm:
_druck_flach(fetch)
return 0
anbieter = args.anbieter or _anbieter_aus_umgebung()
if anbieter is None:
print(
"Abbruch (Kategorisierung): kein LLM-Key gesetzt "
"(OPENROUTER_API_KEY oder ANTHROPIC_API_KEY).\n"
" Vorschlag: einen Key setzen ODER mit --no-llm die belegte Rohliste ausgeben.",
file=sys.stderr,
)
return 2
modell = args.modell
if args.modell_waehlen:
if anbieter != "openrouter":
print(
"Hinweis: --modell-waehlen ist für OpenRouter gedacht; nutze --anbieter openrouter.",
file=sys.stderr,
)
return 2
from .modellauswahl import waehle_modell_interaktiv
modell = waehle_modell_interaktiv()
if not modell:
print("Keine Modellwahl getroffen -- abgebrochen.", file=sys.stderr)
return 2
from .kategorisieren import baue_kategorisierer, kategorisiere
from .uebersicht import rendern
try:
kat = kategorisiere(list(fetch.angebote), baue_kategorisierer(anbieter, modell))
except AbbruchFehler as e:
print(e.als_text(), file=sys.stderr)
return 2
print(rendern(fetch, kat))
return 0
def _liste_modelle(suchbegriff: str) -> int:
from .modelle import lade_modelle, suche, top_free
try:
alle = lade_modelle()
except Exception as e:
print(f"Modelle nicht abrufbar: {e}", file=sys.stderr)
return 2
if suchbegriff:
treffer = suche(alle, suchbegriff)[:20]
print(f"Suche '{suchbegriff}' (max. 20):")
else:
treffer = top_free(alle, 5)
print("Top-5 Free (tool-fähig, Heuristik):")
if not treffer:
print(" (keine Treffer)")
for i, m in enumerate(treffer, 1):
ctx = f"{m.context:,}".replace(",", ".") if m.context else "?"
frei = "FREE" if m.frei else "paid"
warn = "" if m.tools else " ⚠️ kein tool-calling"
print(f" {i}) {m.id} [{frei}, ctx {ctx}]{warn}")
return 0
def _anbieter_aus_umgebung() -> str | None:
if os.environ.get("OPENROUTER_API_KEY"):
return "openrouter"
if os.environ.get("ANTHROPIC_API_KEY"):
return "anthropic"
return None
def _druck_flach(fetch) -> None:
ort = fetch.ort_name or fetch.ort_plz
print(f"# Angebote {ort} (PLZ {fetch.ort_plz}) — Rohliste, unkategorisiert")
print(f"Quellen: {', '.join(fetch.abgedeckte_quellen) or ''}\n")
if not fetch.angebote:
print("_keine Angebote (kein Auffüllen)_")
for a in fetch.angebote:
preis = f"{a.preis:.2f}".replace(".", ",") if a.preis is not None else "Preis fehlt"
marke = f"{a.marke} " if a.marke else ""
menge = f" ({a.menge})" if a.menge else ""
print(f"- {marke}{a.titel}{menge}{preis} @ {a.haendler}")
if fetch.gesehene_haendler:
print("\nBeobachtete Händler: " + ", ".join(fetch.gesehene_haendler))
for h in fetch.hinweise:
print(f"# {h}")
if __name__ == "__main__":
raise SystemExit(main())

87
src/angebote/config.py Normal file
View file

@ -0,0 +1,87 @@
"""Konfiguration -- bewusst getrennt von der Logik.
Produktgruppen, Quellenliste und Orts-Auflösung leben hier, damit das Team sie
ändern kann, ohne Kern- oder Prompt-Code anzufassen.
"""
from __future__ import annotations
# --- Kategorisierung: geschlossene Liste (siehe SKILL angebote-kategorisieren) ---
PRODUKTGRUPPEN: tuple[str, ...] = (
"Obst & Gemüse",
"Fleisch & Wurst",
"Fisch",
"Molkereiprodukte & Eier",
"Brot & Backwaren",
"Grundnahrungsmittel (Nudeln, Reis, Mehl, Konserven)",
"Süßwaren & Snacks",
"Getränke (alkoholfrei)",
"Alkoholische Getränke",
"Tiefkühl",
"Drogerie & Haushalt",
"Non-Food (Kleidung, Spielzeug, Garten, Technik)",
"Sonstiges",
)
# Auffanggruppe, wenn ein Artikel in keine passt ODER das Modell eine Gruppe
# außerhalb der Liste liefert. Muss in PRODUKTGRUPPEN enthalten sein.
FALLBACK_GRUPPE: str = "Sonstiges"
# --- Fetch: bekannte Abdeckungslöcher, die ehrlich ausgewiesen werden ---
# Sentinel-Werte der Quelle für "kein Wert". marktguru setzt z. B. einen
# Platzhalter-Markennamen, wenn keine Marke vorliegt -- der darf NICHT als echte
# Marke durchgereicht werden, sonst behaupten wir einen Beleg, den es nicht gibt.
MARKE_SENTINELS: tuple[str, ...] = ("thisisnobrand123",)
# Hinweis Abdeckung: Welche Händler tatsächlich enthalten sind, ergibt sich aus
# den abgerufenen Daten (datengetrieben), nicht aus einer hartkodierten Annahme.
# Eine pauschale "Discounter X fehlt"-Behauptung wäre unbelegt -- Discounter wie
# Aldi/Lidl können je nach Quelle/Ort sehr wohl enthalten sein.
# Seedbegriffe für query-orientierte Quellen (z. B. marktguru): deren API hat
# kein "alle Angebote"-Browse (leeres q -> 0 Treffer), nur Suche. Diese Liste
# spannt die Abdeckung auf. Sie ist BEWUSST endlich und damit unvollständig --
# was sie nicht trifft, fehlt ehrlich, statt vorgetäuscht zu werden.
SUCHBEGRIFFE: tuple[str, ...] = (
"Obst", "Gemüse", "Apfel", "Banane", "Tomate", "Kartoffel", "Salat",
"Fleisch", "Hähnchen", "Hackfleisch", "Schnitzel", "Wurst", "Schinken",
"Fisch", "Lachs",
"Milch", "Butter", "Käse", "Joghurt", "Quark", "Eier", "Sahne",
"Brot", "Brötchen", "Toast",
"Nudeln", "Reis", "Mehl", "Zucker", "Öl", "Konserve",
"Schokolade", "Süßigkeiten", "Chips", "Kekse",
"Wasser", "Saft", "Cola", "Kaffee", "Tee",
"Bier", "Wein", "Sekt",
"Pizza", "Eis", "Tiefkühl",
"Waschmittel", "Shampoo", "Toilettenpapier", "Zahnpasta",
)
# --- Orts-Auflösung (deterministisch, KEIN LLM) ---
#
# 5-stellige Eingaben gelten direkt als PLZ. Ortsnamen werden NUR über diese
# bewusst kleine, kuratierte Auswahl großer Städte aufgelöst. Ein unbekannter
# Ortsname führt zum ehrlichen Abbruch (Regel 4) mit dem Vorschlag, eine PLZ
# anzugeben -- es wird KEINE PLZ geraten.
ORTSNAME_PLZ: dict[str, str] = {
"berlin": "10115",
"hamburg": "20095",
"münchen": "80331",
"muenchen": "80331",
"köln": "50667",
"koeln": "50667",
"frankfurt": "60311",
"frankfurt am main": "60311",
"stuttgart": "70173",
"düsseldorf": "40213",
"duesseldorf": "40213",
"leipzig": "04109",
"dortmund": "44135",
"essen": "45127",
"bremen": "28195",
"dresden": "01067",
"hannover": "30159",
"nürnberg": "90402",
"nuernberg": "90402",
}

25
src/angebote/fehler.py Normal file
View file

@ -0,0 +1,25 @@
"""Fehler, die den Abbruch statt stiller Drift erzwingen.
`AbbruchFehler` ist Regel 4 in Codeform: Schwelle, Ursache und ein konkreter
Vorschlag werden mitgeführt, damit der Abbruch ein *brauchbares* Ergebnis ist --
nicht nur ein Stacktrace.
"""
from __future__ import annotations
class AbbruchFehler(Exception):
"""Die Datenlage trägt die Anforderung nicht -- bewusster, belegter Abbruch."""
def __init__(self, schwelle: str, ursache: str, vorschlag: str) -> None:
self.schwelle = schwelle
self.ursache = ursache
self.vorschlag = vorschlag
super().__init__(f"{schwelle}: {ursache} -- Vorschlag: {vorschlag}")
def als_text(self) -> str:
return (
f"Abbruch ({self.schwelle}).\n"
f" Ursache: {self.ursache}\n"
f" Vorschlag: {self.vorschlag}"
)

98
src/angebote/fetch.py Normal file
View file

@ -0,0 +1,98 @@
"""Fetch-Orchestrator. Ort rein, belegte Angebote raus. KEIN LLM.
Trennt sauber zwei Fälle, die oberflächlich gleich aussehen:
* KEINE Quelle deckt den Ort ab / Ort nicht auflösbar -> AbbruchFehler (Regel 4).
* Quellen liefen, lieferten aber nichts -> leeres, ehrliches Ergebnis
(kein Auffüllen). Die Übersicht zeigt dann "keine Angebote".
"""
from __future__ import annotations
from .config import ORTSNAME_PLZ
from .fehler import AbbruchFehler
from .modell import Angebot, FetchErgebnis
from .quellen.basis import Ort, QuelleAdapter
def aufloesen_ort(eingabe: str) -> Ort:
"""Löst eine Eingabe (PLZ oder Ortsname) deterministisch zu einem Ort auf.
5-stellige Ziffernfolge -> PLZ. Bekannter Ortsname -> hinterlegte PLZ.
Sonst Abbruch (Regel 4) -- es wird KEINE PLZ geraten.
"""
roh = (eingabe or "").strip()
if roh.isdigit() and len(roh) == 5:
return Ort(plz=roh, name=None)
schluessel = roh.lower()
if schluessel in ORTSNAME_PLZ:
return Ort(plz=ORTSNAME_PLZ[schluessel], name=roh)
raise AbbruchFehler(
schwelle="Ortsauflösung",
ursache=(
f"'{eingabe}' ist weder eine 5-stellige PLZ noch ein hinterlegter Ortsname"
),
vorschlag=(
"eine 5-stellige PLZ angeben (z. B. 60487). Ortsnamen sind nur für "
"eine kleine Auswahl großer Städte hinterlegt -- bewusst wird keine "
"PLZ geraten."
),
)
def hole_angebote(
eingabe: str,
quellen: list[QuelleAdapter] | None = None,
) -> FetchErgebnis:
"""Beschafft belegte Angebote für einen Ort über alle abdeckenden Quellen."""
if quellen is None:
quellen = standard_quellen()
if not quellen:
raise AbbruchFehler(
schwelle="Quellen",
ursache="keine Quelle konfiguriert",
vorschlag="mindestens einen Adapter in standard_quellen() registrieren",
)
ort = aufloesen_ort(eingabe)
abdeckend = [q for q in quellen if q.deckt_ab(ort)]
if not abdeckend:
raise AbbruchFehler(
schwelle="Ortsabdeckung",
ursache=f"keine der {len(quellen)} Quellen deckt PLZ {ort.plz} ab",
vorschlag="andere PLZ in der Nähe versuchen oder weitere Quelle ergänzen",
)
alle: list[Angebot] = []
gelaufen: list[str] = []
hinweise: list[str] = []
for q in abdeckend:
gelaufen.append(q.name)
treffer = q.hole(ort) # AbbruchFehler propagiert bewusst nach oben
if not treffer:
hinweise.append(f"Quelle '{q.name}': 0 Angebote (kein Auffüllen).")
# Ehrliche Abdeckungs-Notiz der Quelle übernehmen, falls vorhanden:
notiz = getattr(q, "abdeckungshinweis", None)
if notiz:
hinweise.append(notiz)
alle.extend(treffer)
gesehene = tuple(sorted({a.haendler for a in alle}))
return FetchErgebnis(
ort_plz=ort.plz,
ort_name=ort.name,
angebote=tuple(alle),
abgedeckte_quellen=tuple(gelaufen),
gesehene_haendler=gesehene,
hinweise=tuple(hinweise),
)
def standard_quellen() -> list[QuelleAdapter]:
"""Registry der aktiven Quellen. Hier werden Adapter ein-/ausgehängt."""
from .quellen.marktguru import MarktguruAdapter
return [MarktguruAdapter()]

View file

@ -0,0 +1,354 @@
"""Kategorisier-Teil -- der EINZIGE Ort mit LLM.
Aufbau bewusst so, dass die harten Regeln im *Code* erzwungen werden, nicht im
Prompt erbeten:
* Dem Modell wird pro Posten nur {id, titel, marke, menge} gezeigt -- nie
Preis/Gültigkeit/Händler. Es kann sie also gar nicht "korrigieren".
* Vom Modell übernommen werden NUR `gruppe` + `unsicher`. Das Original-Angebot
bleibt unverändert (es ist ohnehin frozen).
* Eine Gruppe außerhalb der geschlossenen Liste wird NICHT übernommen ->
Fallback-Gruppe + `unsicher=True`.
* Ein Posten, den das Modell gar nicht beantwortet, wird `unsicher` mit
Fallback-Gruppe -- nicht still einsortiert.
Das LLM steckt hinter `Kategorisierer`; Tests reichen einen Fake herein und
laufen damit ohne Netz und ohne Schlüssel.
"""
from __future__ import annotations
import json
import os
from typing import Protocol
from .config import FALLBACK_GRUPPE, PRODUKTGRUPPEN
from .fehler import AbbruchFehler
from .modell import Angebot, KategorisiertesAngebot
class Kategorisierer(Protocol):
"""Bekommt Posten [{id, titel, marke, menge}], gibt [{id, gruppe, unsicher}]."""
def klassifiziere(self, posten: list[dict]) -> list[dict]: ...
def kategorisiere(
angebote: list[Angebot],
kategorisierer: Kategorisierer,
*,
batch_groesse: int = 25,
fortschritt=None,
) -> list[KategorisiertesAngebot]:
"""Ordnet jedes Angebot einer Produktgruppe zu, ohne Daten zu verändern.
`fortschritt`: optionaler Callback (erledigte_batches, gesamt_batches) --
für Live-Anzeigen (z. B. die Web-UI). Ändert die Logik nicht.
"""
original = {a.angebot_id: a for a in angebote}
ergebnis: dict[str, KategorisiertesAngebot] = {}
import math
gesamt_batches = max(1, math.ceil(len(angebote) / batch_groesse))
for nr, start in enumerate(range(0, len(angebote), batch_groesse), 1):
batch = angebote[start : start + batch_groesse]
posten = [
{
"id": a.angebot_id,
"titel": a.titel,
"marke": a.marke,
"menge": a.menge,
}
for a in batch
]
antworten = kategorisierer.klassifiziere(posten)
for ant in antworten:
aid = ant.get("id")
if aid not in original or aid in ergebnis:
continue # fremde/duplizierte ID ignorieren
gruppe, unsicher = _bereinige_gruppe(ant.get("gruppe"), ant.get("unsicher"))
ergebnis[aid] = KategorisiertesAngebot(
angebot=original[aid], gruppe=gruppe, unsicher=unsicher
)
if fortschritt is not None:
fortschritt(nr, gesamt_batches)
# Posten, die das Modell nicht (gültig) beantwortet hat: ehrlich als
# unsicher mit Fallback markieren -- nicht still einsortieren.
out: list[KategorisiertesAngebot] = []
for a in angebote:
out.append(
ergebnis.get(
a.angebot_id,
KategorisiertesAngebot(angebot=a, gruppe=FALLBACK_GRUPPE, unsicher=True),
)
)
return out
def _bereinige_gruppe(gruppe, unsicher) -> tuple[str, bool]:
"""Erzwingt die geschlossene Liste. Off-list -> Fallback + unsicher."""
flag = bool(unsicher)
if isinstance(gruppe, str) and gruppe in PRODUKTGRUPPEN:
return gruppe, flag
return FALLBACK_GRUPPE, True
# --- echte LLM-Implementierung ----------------------------------------------
_SYSTEM = (
"Du ordnest Supermarkt-Angebote genau EINER Produktgruppe aus einer "
"geschlossenen Liste zu. Erfinde keine neuen Gruppen. Wähle ausschließlich "
"aus dieser Liste:\n- "
+ "\n- ".join(PRODUKTGRUPPEN)
+ "\n\nPasst ein Artikel in keine, nimm 'Sonstiges'. Wenn du dir unsicher "
"bist, setze unsicher=true und gib die wahrscheinlichste Gruppe an -- rate "
"nicht still. Verändere keine Produktnamen. Antworte für JEDEN übergebenen "
"Posten genau einmal über das Tool, identifiziert durch seine id."
)
_TOOL = {
"name": "zuordnungen",
"description": "Gibt für jeden Posten die Produktgruppe und ein Unsicherheits-Flag zurück.",
"input_schema": {
"type": "object",
"properties": {
"zuordnungen": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"gruppe": {"type": "string", "enum": list(PRODUKTGRUPPEN)},
"unsicher": {"type": "boolean"},
},
"required": ["id", "gruppe", "unsicher"],
},
}
},
"required": ["zuordnungen"],
},
}
def _user_text(posten: list[dict]) -> str:
return "Ordne diese Posten zu:\n" + "\n".join(
f"- id={p['id']} | titel={p['titel']} | "
f"marke={p.get('marke')} | menge={p.get('menge')}"
for p in posten
)
def _wartezeit(antwort, versuch: int, basis: float) -> float:
"""Backoff: Retry-After respektieren, sonst exponentiell ab max(basis, 2s)."""
try:
ra = antwort.headers.get("Retry-After")
except Exception:
ra = None
if ra:
try:
return float(ra)
except (TypeError, ValueError):
pass
grund = basis if basis > 0 else 2.0
return grund * (2**versuch)
class AnthropicKategorisierer:
"""Kategorisierer auf Basis der nativen Anthropic-API (Tool-Use)."""
def __init__(self, *, client=None, modell: str = "claude-sonnet-4-6") -> None:
if client is None:
import anthropic
client = anthropic.Anthropic() # liest ANTHROPIC_API_KEY aus der Umgebung
self._client = client
self._modell = modell
def klassifiziere(self, posten: list[dict]) -> list[dict]:
nachricht = self._client.messages.create(
model=self._modell,
max_tokens=2048,
system=[
{"type": "text", "text": _SYSTEM, "cache_control": {"type": "ephemeral"}}
],
tools=[_TOOL],
tool_choice={"type": "tool", "name": "zuordnungen"},
messages=[{"role": "user", "content": _user_text(posten)}],
)
for block in nachricht.content:
if getattr(block, "type", None) == "tool_use":
return list(block.input.get("zuordnungen", []))
return []
class OpenRouterKategorisierer:
"""Kategorisierer über OpenRouter (OpenAI-kompatibles Tool-Calling).
Nutzt nur `requests` -- kein anbieterspezifisches SDK. Liest den Zugang aus
OPENROUTER_API_KEY. Modell-IDs im OpenRouter-Namespace, z. B.
'anthropic/claude-sonnet-4.6'. Bei Transport-/Antwortfehlern: ehrlicher
AbbruchFehler statt stiller Drift.
"""
def __init__(
self,
*,
api_key: str | None = None,
modell: str = "anthropic/claude-sonnet-4.6",
base_url: str = "https://openrouter.ai/api/v1",
session=None,
max_versuche: int = 5,
mindest_abstand_s: float | None = None,
schlafen=None,
) -> None:
self._key = api_key or os.environ.get("OPENROUTER_API_KEY")
if not self._key:
raise AbbruchFehler(
schwelle="openrouter: Zugang",
ursache="OPENROUTER_API_KEY ist nicht gesetzt",
vorschlag="Key setzen ODER --no-llm für die belegte Rohliste verwenden",
)
self._modell = modell
self._base = base_url.rstrip("/")
self._session = session
self._max = max(1, max_versuche)
# Free-Modelle drosseln hart (~20/min). Endet die ID auf ':free' und ist
# kein Abstand vorgegeben, throttlen wir automatisch auf ~20/min.
if mindest_abstand_s is None:
mindest_abstand_s = 3.2 if modell.endswith(":free") else 0.0
self._abstand = mindest_abstand_s
self._letzter = 0.0
# injizierbar für Tests (kein echtes Warten):
if schlafen is None:
import time
schlafen = time.sleep
self._schlafen = schlafen
def _sess(self):
if self._session is None:
import requests
self._session = requests.Session()
return self._session
def _throttle(self) -> None:
if self._abstand <= 0:
return
import time
seit = time.monotonic() - self._letzter
if 0 < seit < self._abstand:
self._schlafen(self._abstand - seit)
self._letzter = time.monotonic()
def klassifiziere(self, posten: list[dict]) -> list[dict]:
tool = {
"type": "function",
"function": {
"name": _TOOL["name"],
"description": _TOOL["description"],
"parameters": _TOOL["input_schema"],
},
}
payload = {
"model": self._modell,
"max_tokens": 2048,
"messages": [
{"role": "system", "content": _SYSTEM},
{"role": "user", "content": _user_text(posten)},
],
"tools": [tool],
"tool_choice": {"type": "function", "function": {"name": _TOOL["name"]}},
}
headers = {
"Authorization": f"Bearer {self._key}",
"Content-Type": "application/json",
"X-Title": "angebote-uebersicht",
}
daten = None
for versuch in range(self._max):
self._throttle()
try:
antwort = self._sess().post(
f"{self._base}/chat/completions",
json=payload,
headers=headers,
timeout=60,
)
except Exception as e:
if versuch + 1 < self._max:
self._schlafen(self._abstand or 2.0)
continue
raise AbbruchFehler(
schwelle="openrouter: Kategorisierung",
ursache=f"chat/completions nicht erreichbar ({e})",
vorschlag="Netz prüfen oder --no-llm verwenden",
) from e
status = getattr(antwort, "status_code", 200)
# 429 (Rate-Limit) / 5xx -> mit Backoff erneut versuchen.
if status == 429 or status >= 500:
if versuch + 1 < self._max:
self._schlafen(_wartezeit(antwort, versuch, self._abstand))
continue
raise AbbruchFehler(
schwelle="openrouter: Rate-Limit",
ursache=f"HTTP {status} auch nach {self._max} Versuchen "
"(Free-Modell zu stark gedrosselt?)",
vorschlag="bezahltes Modell wählen, später erneut, oder --no-llm",
)
try:
antwort.raise_for_status()
daten = antwort.json()
except Exception as e:
raise AbbruchFehler(
schwelle="openrouter: Kategorisierung",
ursache=f"chat/completions fehlgeschlagen ({e})",
vorschlag="Key/Modell/Guthaben prüfen oder --no-llm verwenden",
) from e
break
if isinstance(daten, dict) and daten.get("error"):
raise AbbruchFehler(
schwelle="openrouter: Kategorisierung",
ursache=f"API-Fehler: {daten['error']}",
vorschlag="Key/Modell/Guthaben prüfen oder --no-llm verwenden",
)
try:
aufrufe = daten["choices"][0]["message"]["tool_calls"]
args = json.loads(aufrufe[0]["function"]["arguments"])
return list(args.get("zuordnungen", []))
except (KeyError, IndexError, TypeError, json.JSONDecodeError):
# Modell hat das Tool nicht (sauber) aufgerufen -> leer; die
# kategorisiere()-Logik markiert die Posten dann als unsicher.
return []
_DEFAULT_MODELLE = {
"anthropic": "claude-sonnet-4-6",
"openrouter": "anthropic/claude-sonnet-4.6",
}
def baue_kategorisierer(
anbieter: str, modell: str | None = None, *, api_key: str | None = None
) -> Kategorisierer:
"""Factory: wählt die LLM-Implementierung hinter dem gleichen Protokoll.
`api_key` optional -- ohne wird er aus der Umgebung gelesen (CLI-Fall);
mitgegeben für die Web-UI, falls der Key nicht in der Server-Env steht.
"""
modell = modell or _DEFAULT_MODELLE.get(anbieter)
if anbieter == "openrouter":
return OpenRouterKategorisierer(modell=modell, api_key=api_key)
if anbieter == "anthropic":
return AnthropicKategorisierer(modell=modell)
raise AbbruchFehler(
schwelle="LLM-Anbieter",
ursache=f"unbekannter Anbieter '{anbieter}'",
vorschlag="openrouter oder anthropic wählen",
)

90
src/angebote/modell.py Normal file
View file

@ -0,0 +1,90 @@
"""Datenmodell des Fetch-Teils.
`Angebot` ist bewusst `frozen=True`: Der Kategorisier-Schritt soll die Daten
nicht nur nicht verändern -- er *kann* es nicht. Damit ist die Regel "Daten
sind unantastbar" eine prüfbare Eigenschaft des Typs, keine Bitte.
Es gibt bewusst KEIN Feld `produktgruppe`: Die Trennung von Beschaffung und
Einordnung ist im Datenmodell verankert (siehe SKILL angebote-fetch).
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import date, datetime
@dataclass(frozen=True)
class Angebot:
"""Ein normalisiertes, belegtes Angebot.
Pflichtfelder (`haendler`, `quelle`, `abgerufen_am`) belegen die Herkunft.
Optionale Felder sind `None`, wenn die Quelle sie nicht eindeutig hergibt --
niemals geraten.
"""
titel: str
haendler: str
quelle: str
abgerufen_am: datetime
marke: str | None = None
preis: float | None = None
grundpreis: str | None = None
menge: str | None = None
gueltig_von: date | None = None
gueltig_bis: date | None = None
# Stabile ID für das Zurückmappen nach der Kategorisierung. Wird aus
# belegten Feldern abgeleitet, nicht erfunden.
angebot_id: str = ""
def __post_init__(self) -> None:
if not self.titel or not self.titel.strip():
raise ValueError("Angebot ohne Titel ist nicht belegbar")
if not self.haendler or not self.haendler.strip():
raise ValueError("Angebot ohne Händler verletzt 'nur Belegtes'")
if not self.quelle or not self.quelle.strip():
raise ValueError("Angebot ohne Quelle verletzt 'nur Belegtes'")
if not self.angebot_id:
# frozen -> über object.__setattr__ setzen
object.__setattr__(self, "angebot_id", self._ableiten_id())
def _ableiten_id(self) -> str:
import hashlib
roh = "|".join(
[
self.haendler,
self.quelle,
self.titel,
str(self.preis),
str(self.gueltig_von),
str(self.gueltig_bis),
]
)
return hashlib.sha1(roh.encode("utf-8")).hexdigest()[:16]
@dataclass(frozen=True)
class KategorisiertesAngebot:
"""Ein Angebot plus die vom LLM vergebene Gruppe.
Das Original-`Angebot` bleibt unverändert eingebettet -- übernommen werden
aus dem Modell ausschließlich `gruppe` und `unsicher`.
"""
angebot: Angebot
gruppe: str
unsicher: bool = False
@dataclass(frozen=True)
class FetchErgebnis:
"""Ergebnis des Fetch-Teils: belegte Angebote + ehrliche Abdeckungslage."""
ort_plz: str
ort_name: str | None
angebote: tuple[Angebot, ...]
abgedeckte_quellen: tuple[str, ...]
# Datengetrieben: die Händler, die tatsächlich in den Angeboten vorkamen.
gesehene_haendler: tuple[str, ...] = field(default_factory=tuple)
hinweise: tuple[str, ...] = field(default_factory=tuple)

View file

@ -0,0 +1,100 @@
"""Interaktive Modell-Auswahl für den OpenRouter-Kategorisierer.
Zeigt die aktuell besten Top-5-Free-Modelle, erlaubt Suche und Aktualisieren
und gibt die gewählte Modell-ID zurück. I/O ist injizierbar (`eingabe`/`ausgabe`),
damit der Ablauf ohne TTY testbar ist.
Befehle im Picker:
[1..N] Nummer aus der aktuellen Liste wählen
s TEXT nach TEXT suchen (über alle Modelle)
f zurück zu den Top-5 Free
u Liste neu von OpenRouter laden ("Aktualisieren")
q abbrechen
"""
from __future__ import annotations
from typing import Callable
from .modelle import ModellInfo, lade_modelle, suche, top_free
def _format(modelle: list[ModellInfo], titel: str) -> str:
zeilen = [titel]
for i, m in enumerate(modelle, 1):
ctx = f"{m.context:,}".replace(",", ".") if m.context else "?"
frei = "FREE" if m.frei else "paid"
warn = "" if m.tools else " ⚠️ KEIN tool-calling (Kategorisierung bliebe leer)"
zeilen.append(f" {i}) {m.id} [{frei}, ctx {ctx}]{warn}")
if not modelle:
zeilen.append(" (keine Treffer)")
return "\n".join(zeilen)
def waehle_modell_interaktiv(
*,
session=None,
eingabe: Callable[[str], str] = input,
ausgabe: Callable[[str], None] = print,
) -> str | None:
"""Führt den Auswahldialog und gibt die gewählte Modell-ID zurück (oder None)."""
ausgabe("Lade OpenRouter-Modelle …")
alle = lade_modelle(session)
aktuell = top_free(alle, 5)
titel = "Top-5 Free (tool-fähig, Heuristik siehe modelle.py):"
hilfe = (
"Befehle: [Nummer] wählen · [s TEXT] suchen · [f] Top-5 Free · "
"[u] aktualisieren · [q] abbrechen"
)
while True:
ausgabe(_format(aktuell, titel))
ausgabe(hilfe)
try:
roh = eingabe("> ").strip()
except EOFError:
return None
if not roh:
continue
befehl = roh.lower()
if befehl in ("q", "quit", "abbrechen"):
return None
if befehl in ("u", "update"):
ausgabe("Aktualisiere …")
alle = lade_modelle(session)
aktuell = top_free(alle, 5)
titel = "Top-5 Free (aktualisiert):"
continue
if befehl in ("f", "free"):
aktuell = top_free(alle, 5)
titel = "Top-5 Free (tool-fähig, Heuristik):"
continue
if befehl.startswith("s ") or befehl == "s":
begriff = roh[1:].strip()
aktuell = suche(alle, begriff)[:15]
titel = f"Suche '{begriff}' (max. 15):"
continue
if roh.isdigit():
idx = int(roh) - 1
if 0 <= idx < len(aktuell):
gewaehlt = aktuell[idx]
if not gewaehlt.tools:
ausgabe(
"Achtung: Modell kann kein Tool-Calling — die Kategorisierung "
"würde leer bleiben. Trotzdem wählen? [j/N]"
)
if eingabe("> ").strip().lower() not in ("j", "ja", "y"):
continue
ausgabe(f"Gewählt: {gewaehlt.id}")
return gewaehlt.id
ausgabe("Nummer außerhalb der Liste.")
continue
ausgabe("Eingabe nicht verstanden.")

115
src/angebote/modelle.py Normal file
View file

@ -0,0 +1,115 @@
"""OpenRouter-Modell-Discovery für die Kategorisierer-Auswahl.
Rein deterministisch (KEIN LLM): zieht die Modellliste live von OpenRouter,
erkennt frei nutzbare und tool-fähige Modelle und sortiert sie nach einer
DOKUMENTIERTEN Heuristik.
Ehrlichkeit zu "beste": Die API liefert keine Qualitätsmetrik. "beste" ist hier
deshalb kein Benchmark, sondern eine nachvollziehbare Präferenz bekannter,
starker Modellfamilien -- und IMMER gefiltert auf das, was gerade real
verfügbar UND tool-fähig ist (sonst liefert die Kategorisierung leer).
"""
from __future__ import annotations
from dataclasses import dataclass
MODELS_URL = "https://openrouter.ai/api/v1/models"
# Dokumentierte Präferenz (Meinung, kein Benchmark). Reihenfolge = Rang.
# Modelle, die hier matchen, gelten als "stärker"; der Rest folgt nach
# Kontextgröße. Bei Bedarf hier anpassen -- die Liste lebt in der Config-Ebene,
# nicht in der Logik.
PRAEFERENZ: tuple[str, ...] = (
"kimi-k2",
"qwen3-next",
"llama-3.3-70b",
"gpt-oss-120b",
"glm-4.5",
"qwen3-coder",
"nemotron-3-super",
"gemma-4-31b",
"qwen3",
"llama",
"gemma",
"nemotron",
"gpt-oss",
"mistral",
)
@dataclass(frozen=True)
class ModellInfo:
id: str
name: str
context: int | None
frei: bool
tools: bool
def _als_float(x) -> float:
try:
return float(x)
except (TypeError, ValueError):
return 1.0 # unbekannt -> als "nicht frei" werten, nicht raten
def parse_modelle(daten: list[dict]) -> list[ModellInfo]:
out: list[ModellInfo] = []
for m in daten:
p = m.get("pricing") or {}
frei = _als_float(p.get("prompt")) == 0 and _als_float(p.get("completion")) == 0
params = m.get("supported_parameters") or []
out.append(
ModellInfo(
id=m.get("id", ""),
name=m.get("name") or m.get("id", ""),
context=m.get("context_length"),
frei=frei,
tools="tools" in params,
)
)
return out
def lade_modelle(session=None) -> list[ModellInfo]:
"""Zieht die Modellliste live (= 'aktualisieren'). Kein Key nötig."""
sess = session
if sess is None:
import requests
sess = requests
antwort = sess.get(MODELS_URL, timeout=30)
antwort.raise_for_status()
return parse_modelle(antwort.json().get("data", []))
def _rang(mi: ModellInfo) -> tuple:
low = mi.id.lower()
for i, schluessel in enumerate(PRAEFERENZ):
if schluessel in low:
return (0, i, -(mi.context or 0))
return (1, 0, -(mi.context or 0))
def top_free(modelle: list[ModellInfo], n: int = 5, nur_tools: bool = True) -> list[ModellInfo]:
"""Die n besten frei nutzbaren Modelle (standardmäßig nur tool-fähige)."""
kandidaten = [m for m in modelle if m.frei and (m.tools or not nur_tools)]
return sorted(kandidaten, key=_rang)[:n]
def suche(
modelle: list[ModellInfo],
begriff: str,
*,
nur_frei: bool = False,
nur_tools: bool = False,
) -> list[ModellInfo]:
"""Filtert nach Teilstring in id/name; optional auf frei/tool-fähig."""
b = (begriff or "").lower().strip()
res = [m for m in modelle if b in m.id.lower() or b in m.name.lower()]
if nur_frei:
res = [m for m in res if m.frei]
if nur_tools:
res = [m for m in res if m.tools]
return sorted(res, key=_rang)

View file

@ -0,0 +1 @@
"""Quellen-Adapter. Ein Adapter pro Quelle, gleiche Schnittstelle (basis.py)."""

View file

@ -0,0 +1,41 @@
"""Adapter-Schnittstelle für Quellen.
Vertrag: rein = `Ort`, raus = `list[Angebot]`. Jeder Adapter kapselt genau eine
Quelle; die Kernlogik (fetch.py) kennt nur diese Schnittstelle, nicht die
einzelne Quelle. So lassen sich Quellen austauschen, ohne den Kern anzufassen.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Protocol, runtime_checkable
from ..modell import Angebot
@dataclass(frozen=True)
class Ort:
"""Aufgelöster Ort. `plz` ist Pflicht -- ein Adapter filtert darüber."""
plz: str
name: str | None = None
@runtime_checkable
class QuelleAdapter(Protocol):
"""Eine Angebotsquelle.
`name` -- Anzeigename der Quelle.
`deckt_ab` -- ob die Quelle den Ort bedienen kann (entscheidet mit über
Regel 4: deckt KEINE Quelle den Ort ab -> Abbruch).
`hole` -- liefert belegte Angebote für den Ort. Liefert die Quelle
nichts, ist das ein leeres Ergebnis -- KEIN Auffüllen.
Kann die Quelle den Ortsbezug nicht verifizieren oder den
nötigen Zugang nicht herstellen, wirft sie AbbruchFehler.
"""
name: str
def deckt_ab(self, ort: Ort) -> bool: ...
def hole(self, ort: Ort) -> list[Angebot]: ...

View file

@ -0,0 +1,427 @@
"""marktguru-Adapter -- erste echte Quelle. REIN DETERMINISTISCH, kein LLM.
Die API ist öffentlich dokumentiert nachvollziehbar (Reverse-Engineering durch
mehrere Community-Projekte):
Basis : https://api.marktguru.de/api/v1
Endpoint : offers/search
Parameter : as=web, q=<suche>, zipCode=<PLZ>, limit, offset
Header : x-apikey, x-clientkey -- werden aus dem in der Startseite
eingebetteten <script type="application/json">-Config-Block gelesen
(Felder apiKey / clientKey), oder per ENV überschrieben.
Felder : offer.product.name, offer.brand.name, offer.price,
offer.advertisers[].name, offer.validityDates[].from / .to,
offer.volume + offer.quantity + offer.unit.name -> Packungsmenge,
offer.referencePrice (numerisch, je unit.name) -> Grundpreis.
Anmerkung Ortsbezug/Abdeckung: offers/search ist QUERY-orientiert -- ein leeres
q liefert 0 Treffer, kein "alle Angebote"-Browse. Eine vollständige
ortsweite Aufzählung gibt die API nicht her; der Adapter fragt deshalb die in
SUCHBEGRIFFE hinterlegten Kategorie-Seedbegriffe ab und meldet diese
Teilabdeckung ehrlich, statt Vollständigkeit vorzutäuschen.
Ortsbezug: `zipCode` ist ein EXPLIZITER Query-Parameter -- der Ortsfilter steckt
also nicht in Session/Cookie. Der Adapter belegt den Filter über den
mitgesendeten `zipCode` und bricht ab (Regel 4), wenn er ihn nicht herstellen
kann. Lässt sich der Zugang (Schlüssel) nicht herstellen -> Abbruch, kein
geratener Schlüssel.
"""
from __future__ import annotations
import json
import os
import re
from datetime import date, datetime
from pathlib import Path
from urllib.robotparser import RobotFileParser
from ..fehler import AbbruchFehler
from ..modell import Angebot
from .basis import Ort
_HOMEPAGE = "https://www.marktguru.de"
_API_BASIS = "https://api.marktguru.de/api/v1"
_OFFERS_SEARCH = f"{_API_BASIS}/offers/search"
_USER_AGENT = "angebote-uebersicht/0.1 (+kontaktloser, robots-treuer Aggregator)"
_CONFIG_RE = re.compile(
r'<script[^>]*type="application/json"[^>]*>(.*?)</script>', re.DOTALL
)
class MarktguruAdapter:
"""Adapter für marktguru.de. Erfüllt das QuelleAdapter-Protokoll."""
name = "marktguru"
def __init__(
self,
*,
suchbegriffe: tuple[str, ...] | None = None,
limit: int = 100,
cache_dir: str | Path | None = None,
session=None,
) -> None:
from ..config import SUCHBEGRIFFE
self.suchbegriffe = tuple(suchbegriffe) if suchbegriffe is not None else SUCHBEGRIFFE
self.limit = limit
self.cache_dir = Path(cache_dir) if cache_dir else Path(".cache/marktguru")
self._session = session # für Tests injizierbar
self._apikey: str | None = None
self._clientkey: str | None = None
# Ehrliche Abdeckungs-Notiz, vom Orchestrator eingesammelt:
self.abdeckungshinweis = (
f"marktguru: query-basierte Teilabdeckung über {len(self.suchbegriffe)} "
"Seedbegriffe -- keine vollständige Aufzählung aller Angebote."
)
# -- QuelleAdapter ----------------------------------------------------
def deckt_ab(self, ort: Ort) -> bool:
# marktguru filtert bundesweit über PLZ. Eine 5-stellige PLZ gilt als
# abgedeckt; ob konkret Angebote vorliegen, entscheidet erst `hole`.
return bool(ort.plz) and ort.plz.isdigit() and len(ort.plz) == 5
def hole(self, ort: Ort) -> list[Angebot]:
if not self.deckt_ab(ort):
raise AbbruchFehler(
schwelle="marktguru: PLZ-Filter",
ursache=f"'{ort.plz}' ist keine gültige 5-stellige PLZ",
vorschlag="eine 5-stellige PLZ angeben (z. B. 60487)",
)
roh, abgerufen_am = self._hole_roh(ort)
angebote = self._parse(roh, ort, abgerufen_am)
# Ortsbezug verifizieren: Es gab Treffer, aber der Filter ist nur dann
# belegt, wenn wir ihn auch wirklich mitgesendet haben (siehe quelle).
# Trefferzahl 0 ist KEIN Fehler -> leeres Ergebnis (kein Auffüllen).
return angebote
# -- intern -----------------------------------------------------------
def _sess(self):
if self._session is None:
import requests # lokal, damit der Fetch-Teil ohne Netz importierbar bleibt
self._session = requests.Session()
self._session.headers.update({"User-Agent": _USER_AGENT})
return self._session
def _robots_pruefen(self, url: str) -> tuple[bool, str]:
"""Prüft robots.txt. Gibt (erlaubt, ehrliche_ursache_falls_nicht) zurück.
robots.txt wird über DENSELBEN Transport wie der eigentliche Abruf
geholt (requests, certifi-gestützt) -- damit ein erfolgreicher Abruf
und ein erfolgreicher robots-Check dieselbe Vertrauensbasis haben.
Wichtig für die Ehrlichkeit des Abbruchs: 'nicht prüfbar' (Netz-/SSL-/
HTTP-Fehler) ist NICHT dasselbe wie 'durch robots.txt untersagt'. Beide
führen konservativ zum Nicht-Abruf, aber die genannte Ursache muss
zutreffen -- es wird keine Sperre behauptet, die nicht belegt ist.
Fehlt robots.txt (404), gilt der Pfad als erlaubt.
"""
from urllib.parse import urlsplit
teile = urlsplit(url)
robots_url = f"{teile.scheme}://{teile.netloc}/robots.txt"
try:
antwort = self._sess().get(robots_url, timeout=20)
except Exception as e:
return False, f"robots.txt nicht prüfbar ({robots_url}: {e})"
if antwort.status_code == 404:
return True, ""
if antwort.status_code != 200:
return False, f"robots.txt nicht prüfbar ({robots_url}: HTTP {antwort.status_code})"
rp = RobotFileParser()
rp.parse(antwort.text.splitlines())
if rp.can_fetch(_USER_AGENT, url):
return True, ""
return False, f"durch robots.txt untersagt ({robots_url})"
def _schluessel(self) -> tuple[str, str]:
if self._apikey and self._clientkey:
return self._apikey, self._clientkey
env_api = os.environ.get("MARKTGURU_APIKEY")
env_client = os.environ.get("MARKTGURU_CLIENTKEY")
if env_api and env_client:
self._apikey, self._clientkey = env_api, env_client
return self._apikey, self._clientkey
erlaubt, grund = self._robots_pruefen(_HOMEPAGE)
if not erlaubt:
raise AbbruchFehler(
schwelle="marktguru: Startseite nicht abrufbar",
ursache=grund,
vorschlag="Schlüssel per ENV MARKTGURU_APIKEY/MARKTGURU_CLIENTKEY setzen",
)
try:
antwort = self._sess().get(_HOMEPAGE, timeout=20)
antwort.raise_for_status()
html = antwort.text
except Exception as e: # Netz-/HTTP-Fehler
raise AbbruchFehler(
schwelle="marktguru: Zugang",
ursache=f"Startseite nicht abrufbar ({e})",
vorschlag="Netzverbindung prüfen oder Schlüssel per ENV setzen",
) from e
api, client = self._schluessel_aus_html(html)
if not api or not client:
raise AbbruchFehler(
schwelle="marktguru: Zugang",
ursache="apiKey/clientKey nicht im Config-Block der Startseite gefunden "
"(Seitenstruktur geändert?)",
vorschlag="Schlüssel per ENV MARKTGURU_APIKEY/MARKTGURU_CLIENTKEY setzen",
)
self._apikey, self._clientkey = api, client
return api, client
@staticmethod
def _schluessel_aus_html(html: str) -> tuple[str | None, str | None]:
for block in _CONFIG_RE.findall(html):
try:
cfg = json.loads(block)
except json.JSONDecodeError:
continue
api, client = _suche_keys(cfg)
if api and client:
return api, client
return None, None
def _cache_pfad(self, ort: Ort, abgerufen_am: datetime) -> Path:
jahr, woche, _ = abgerufen_am.isocalendar()
return self.cache_dir / f"{ort.plz}_{jahr}-W{woche:02d}.json"
def _hole_roh(self, ort: Ort) -> tuple[list[dict], datetime]:
"""Aggregiert die rohen Offer-Objekte über alle Seedbegriffe (dedupliziert).
Cache pro PLZ/Kalenderwoche; bei Treffer wird nicht erneut gezogen.
Ein einzelner Suchbegriff ohne Treffer ist KEIN Fehler -- nur ein
leerer Beitrag. Bricht hingegen der Zugang/HTTP, gilt Regel 4.
"""
jetzt = datetime.now()
treffer = self._cache_pfad(ort, jetzt)
if treffer.exists():
gespeichert = json.loads(treffer.read_text(encoding="utf-8"))
return gespeichert["results"], datetime.fromisoformat(
gespeichert["abgerufen_am"]
)
erlaubt, grund = self._robots_pruefen(_OFFERS_SEARCH)
if not erlaubt:
raise AbbruchFehler(
schwelle="marktguru: API nicht abrufbar",
ursache=grund,
vorschlag="andere Quelle wählen, Netz/Zertifikate prüfen oder Betreiber kontaktieren",
)
api, client = self._schluessel()
headers = {"x-apikey": api, "x-clientkey": client}
gesehen: set[str] = set()
aggregiert: list[dict] = []
for begriff in self.suchbegriffe:
for offer in self._suche_eine(begriff, ort, headers):
oid = str(offer.get("id") or "")
schluessel = oid or json.dumps(offer, sort_keys=True)[:120]
if schluessel in gesehen:
continue
gesehen.add(schluessel)
aggregiert.append(offer)
self.cache_dir.mkdir(parents=True, exist_ok=True)
treffer.write_text(
json.dumps({"abgerufen_am": jetzt.isoformat(), "results": aggregiert}),
encoding="utf-8",
)
return aggregiert, jetzt
def _suche_eine(self, begriff: str, ort: Ort, headers: dict) -> list[dict]:
params = {
"as": "web",
"q": begriff,
"limit": str(self.limit),
"offset": "0",
"zipCode": ort.plz, # <-- belegter Ortsfilter
}
try:
antwort = self._sess().get(
_OFFERS_SEARCH, params=params, headers=headers, timeout=30
)
antwort.raise_for_status()
daten = antwort.json()
except Exception as e:
raise AbbruchFehler(
schwelle="marktguru: Abruf",
ursache=f"offers/search fehlgeschlagen für '{begriff}' ({e})",
vorschlag="später erneut versuchen, PLZ prüfen, ggf. Schlüssel erneuern",
) from e
results = daten.get("results")
if results is None:
raise AbbruchFehler(
schwelle="marktguru: Antwortform",
ursache="Antwort enthält kein 'results' (API-Form geändert?)",
vorschlag="Parser an neue Antwortstruktur anpassen",
)
return results
def _parse(
self, results: list[dict], ort: Ort, abgerufen_am: datetime
) -> list[Angebot]:
angebote: list[Angebot] = []
for eintrag in results:
offer = eintrag.get("offer", eintrag) # je nach Antwortform
titel = _pfad(offer, "product", "name")
if not titel:
# Ohne belegten Titel ist der Eintrag nicht verwertbar -> weglassen,
# NICHT mit Platzhalter füllen.
continue
haendler = _erster_name(offer.get("advertisers")) or "unbekannt (marktguru)"
offer_id = str(offer.get("id") or "")
einheit = _pfad(offer, "unit", "name")
angebote.append(
Angebot(
titel=titel,
marke=_marke(_pfad(offer, "brand", "name")),
preis=_als_float(offer.get("price")),
grundpreis=_grundpreis(offer.get("referencePrice"), einheit),
menge=_menge(offer.get("quantity"), offer.get("volume"), einheit),
gueltig_von=_erstes_datum(offer.get("validityDates"), "from"),
gueltig_bis=_erstes_datum(offer.get("validityDates"), "to"),
haendler=haendler,
# quelle belegt zugleich den Ortsfilter (zipCode):
quelle=f"marktguru:offers/search?zipCode={ort.plz}#offer={offer_id}",
abgerufen_am=abgerufen_am,
)
)
return angebote
# --- kleine, reine Hilfsfunktionen (keine Daten-Reparatur, nur Auslesen) -----
def _suche_keys(obj) -> tuple[str | None, str | None]:
"""Findet apiKey/clientKey rekursiv im Config-Objekt der Startseite."""
api = client = None
if isinstance(obj, dict):
for k, v in obj.items():
kl = k.lower()
if kl == "apikey" and isinstance(v, str):
api = v
elif kl == "clientkey" and isinstance(v, str):
client = v
else:
a, c = _suche_keys(v)
api = api or a
client = client or c
elif isinstance(obj, list):
for el in obj:
a, c = _suche_keys(el)
api = api or a
client = client or c
return api, client
def _pfad(obj: dict, *schluessel: str) -> str | None:
cur = obj
for s in schluessel:
if not isinstance(cur, dict):
return None
cur = cur.get(s)
if isinstance(cur, str) and cur.strip():
return cur
return None
def _marke(wert: str | None) -> str | None:
"""Filtert den Sentinel der Quelle für 'keine Marke' heraus."""
from ..config import MARKE_SENTINELS
if wert is None or wert in MARKE_SENTINELS:
return None
return wert
def _erster_name(liste) -> str | None:
if isinstance(liste, list):
for el in liste:
if isinstance(el, dict) and isinstance(el.get("name"), str):
return el["name"]
return None
def _als_float(wert) -> float | None:
if isinstance(wert, (int, float)):
return float(wert)
return None # nicht eindeutig -> fehlend, NICHT geraten
def _zahl(wert) -> str | None:
"""Formatiert eine Zahl ohne überflüssige Nullen, mit Komma. 0.25 -> '0,25'."""
if not isinstance(wert, (int, float)):
return None
text = f"{wert:.3f}".rstrip("0").rstrip(".")
return text.replace(".", ",")
def _grundpreis(wert, einheit: str | None) -> str | None:
"""Grundpreis aus belegten Feldern komponiert -- nichts umgerechnet/erfunden.
marktguru liefert referencePrice numerisch (Preis je `unit.name`). Aus den
zwei belegten Feldern wird eine lesbare, aber unveränderte Darstellung.
"""
if isinstance(wert, (int, float)) and einheit:
return f"{wert:.2f} €/{einheit}".replace(".", ",")
if isinstance(wert, str) and wert.strip():
return wert # Falls eine Quelle es doch als String liefert: roh.
return None
def _menge(quantity, volume, einheit: str | None) -> str | None:
"""Packungsmenge aus quantity × volume + Einheit -- belegt zusammengesetzt."""
vol = _zahl(volume)
if not vol or not einheit:
return None
if isinstance(quantity, (int, float)) and quantity and quantity != 1:
return f"{_zahl(quantity)} × {vol} {einheit}"
return f"{vol} {einheit}"
def _erstes_datum(liste, feld: str) -> date | None:
if isinstance(liste, list):
for el in liste:
if isinstance(el, dict) and el.get(feld):
return _parse_datum(el[feld])
return None
def _parse_datum(roh) -> date | None:
"""Gültigkeits-Datum aus ISO-Zeitstempel.
Zeitzonen-korrekt: marktguru liefert UTC ('...22:00:00Z'). Für einen
Einkäufer in Deutschland zählt das LOKALE Datum -- sonst entsteht ein
Off-by-one (22:00Z = 00:00 lokal am Folgetag). Daher Umrechnung nach
Europe/Berlin, bevor das Datum genommen wird.
"""
if not isinstance(roh, str):
return None
text = roh.replace("Z", "+00:00")
try:
dt = datetime.fromisoformat(text)
except ValueError:
try:
return date.fromisoformat(text[:10])
except ValueError:
return None # nicht parsebar -> fehlend, nicht geraten
if dt.tzinfo is not None:
try:
from zoneinfo import ZoneInfo
dt = dt.astimezone(ZoneInfo("Europe/Berlin"))
except Exception:
pass # ohne tz-Daten: belegtes UTC-Datum statt Abbruch
return dt.date()

137
src/angebote/uebersicht.py Normal file
View file

@ -0,0 +1,137 @@
"""Gruppierung + Rendering der fertigen Übersicht.
Regeln, die auch hier sichtbar bleiben:
* Reihenfolge = feste Produktgruppen-Liste (config), nicht "nach Häufigkeit".
* Leere Gruppen werden als "keine Angebote" gezeigt -- nicht weggelassen,
nicht aufgefüllt.
* Unsicher zugeordnete Artikel sind markiert (Mensch kann nachsehen).
* Bekannte Abdeckungslücken (z. B. Aldi/Lidl) werden ehrlich genannt.
"""
from __future__ import annotations
from .config import PRODUKTGRUPPEN
from .modell import FetchErgebnis, KategorisiertesAngebot
def gruppieren(
kategorisiert: list[KategorisiertesAngebot],
) -> dict[str, list[KategorisiertesAngebot]]:
"""Gruppiert in der festen Reihenfolge der Produktgruppen-Config."""
gruppen: dict[str, list[KategorisiertesAngebot]] = {g: [] for g in PRODUKTGRUPPEN}
for ka in kategorisiert:
# _bereinige_gruppe garantiert: ka.gruppe ist in PRODUKTGRUPPEN.
gruppen[ka.gruppe].append(ka)
return gruppen
def _angebot_dict(ka: KategorisiertesAngebot) -> dict:
a = ka.angebot
return {
"titel": a.titel,
"marke": a.marke,
"preis": a.preis,
"grundpreis": a.grundpreis,
"menge": a.menge,
"haendler": a.haendler,
"gueltig_von": a.gueltig_von.isoformat() if a.gueltig_von else None,
"gueltig_bis": a.gueltig_bis.isoformat() if a.gueltig_bis else None,
"quelle": a.quelle,
"unsicher": ka.unsicher,
}
def als_struktur(
fetch: FetchErgebnis,
kategorisiert: list[KategorisiertesAngebot],
) -> dict:
"""Strukturierte Ausgabe für die Web-UI -- dieselben belegten Felder wie der
Markdown-Renderer, nur als JSON-fähiges dict. Leere Gruppen bleiben enthalten
(als leere Liste) -- kein Weglassen, kein Auffüllen."""
gruppen = gruppieren(kategorisiert)
return {
"ort_plz": fetch.ort_plz,
"ort_name": fetch.ort_name,
"anzahl": len(kategorisiert),
"unsicher": sum(1 for k in kategorisiert if k.unsicher),
"quellen": list(fetch.abgedeckte_quellen),
"haendler": list(fetch.gesehene_haendler),
"hinweise": list(fetch.hinweise),
"gruppen": [
{
"name": g,
"anzahl": len(gruppen[g]),
"angebote": [_angebot_dict(ka) for ka in gruppen[g]],
}
for g in PRODUKTGRUPPEN
],
}
def rendern(
fetch: FetchErgebnis,
kategorisiert: list[KategorisiertesAngebot],
) -> str:
gruppen = gruppieren(kategorisiert)
ort = fetch.ort_name or fetch.ort_plz
zeilen: list[str] = []
zeilen.append(f"# Angebote {ort} (PLZ {fetch.ort_plz})")
zeilen.append("")
zeilen.append(
f"Quellen: {', '.join(fetch.abgedeckte_quellen) or ''} · "
f"{len(kategorisiert)} Angebote"
)
zeilen.append("")
for gruppe in PRODUKTGRUPPEN:
eintraege = gruppen[gruppe]
zeilen.append(f"## {gruppe}")
if not eintraege:
zeilen.append("_keine Angebote_")
zeilen.append("")
continue
for ka in eintraege:
zeilen.append(_zeile(ka))
zeilen.append("")
zeilen.append("---")
if fetch.gesehene_haendler:
zeilen.append(
"**Beobachtete Händler (datengetrieben, belegt):** "
+ ", ".join(fetch.gesehene_haendler)
+ "."
)
zeilen.append(
"_Abdeckung = was die abgefragten Quellen für diesen Ort hergeben; "
"Fehlen eines Händlers belegt keine Lücke, sondern nur 'diese Woche/"
"Abfrage kein Treffer'._"
)
for h in fetch.hinweise:
zeilen.append(f"_{h}_")
return "\n".join(zeilen).rstrip() + "\n"
def _zeile(ka: KategorisiertesAngebot) -> str:
a = ka.angebot
teile: list[str] = []
if a.marke:
teile.append(f"**{a.marke}** {a.titel}")
else:
teile.append(f"**{a.titel}**")
if a.menge:
teile.append(f"({a.menge})")
if a.preis is not None:
teile.append(f"{a.preis:.2f}".replace(".", ","))
else:
teile.append("— Preis fehlt")
if a.grundpreis:
teile.append(f"[{a.grundpreis}]")
if a.gueltig_von or a.gueltig_bis:
von = a.gueltig_von.isoformat() if a.gueltig_von else "?"
bis = a.gueltig_bis.isoformat() if a.gueltig_bis else "?"
teile.append(f"gültig {von}{bis}")
teile.append(f"@ {a.haendler}")
if ka.unsicher:
teile.append("⚠️ unsicher")
return "- " + " ".join(teile)

147
src/angebote/web.py Normal file
View file

@ -0,0 +1,147 @@
"""FastAPI-Web-UI -- dünne Schicht über den bestehenden Modulen.
Der Schnitt bleibt unangetastet: dieser Server ruft `fetch` (deterministisch)
und `kategorisieren` (LLM) auf, vermischt aber nichts. Die UI ist reine
Präsentation; alle harten Regeln (kein Auffüllen, nur Belegtes, Abbruch statt
Drift) leben weiter in den darunterliegenden Modulen.
Start:
OPENROUTER_API_KEY=... uvicorn angebote.web:app --port 8000
(aus dem src/-Verzeichnis, oder mit PYTHONPATH=src)
"""
from __future__ import annotations
import threading
import uuid
from pathlib import Path
from fastapi import FastAPI, HTTPException
from fastapi.responses import HTMLResponse
from .fehler import AbbruchFehler
app = FastAPI(title="Angebots-Übersicht")
_HTML = (Path(__file__).parent / "web_static" / "index.html").read_text("utf-8")
# In-memory Job-Store. Schlicht gehalten -- ein lokales Single-User-Werkzeug.
_jobs: dict[str, dict] = {}
_jobs_lock = threading.Lock()
# Ergebnis-Cache: identischer Lauf (PLZ, Modell, no_llm) kommt sofort, ohne
# erneute LLM-Calls. Im Geist des Projekt-Cachings. In-memory, pro Serverlauf.
_ergebnis_cache: dict[tuple, dict] = {}
@app.get("/", response_class=HTMLResponse)
def index() -> str:
return _HTML
@app.get("/api/modelle")
def api_modelle(q: str = "") -> list[dict]:
"""Modell-Liste für das Dropdown: Suche oder Top-Free. 'Aktualisieren' = neu rufen."""
from .modelle import lade_modelle, suche, top_free
try:
alle = lade_modelle()
except Exception as e: # Netz/SSL -> ehrlich melden, nicht raten
raise HTTPException(status_code=502, detail=f"Modelle nicht abrufbar: {e}")
treffer = suche(alle, q) if q else top_free(alle, 8)
return [
{"id": m.id, "frei": m.frei, "tools": m.tools, "context": m.context}
for m in treffer[:25]
]
@app.post("/api/lauf")
def api_lauf(req: dict) -> dict:
plz = (req.get("plz") or "").strip()
if not plz:
raise HTTPException(status_code=400, detail="PLZ fehlt")
modell = req.get("modell") or None
no_llm = bool(req.get("no_llm"))
job_id = uuid.uuid4().hex[:12]
# Cache-Treffer? Dann sofort als fertiger Job ausliefern, kein neuer Lauf.
treffer = _ergebnis_cache.get((plz, modell, no_llm))
if treffer is not None:
with _jobs_lock:
_jobs[job_id] = {
"status": "fertig", "phase": "cache", "done": 0, "total": 0,
"ergebnis": treffer, "fehler": None,
}
return {"job_id": job_id, "cache": True}
with _jobs_lock:
_jobs[job_id] = {
"status": "laufend",
"phase": "fetch",
"done": 0,
"total": 0,
"ergebnis": None,
"fehler": None,
}
t = threading.Thread(
target=_run_job,
args=(
job_id,
plz,
modell,
req.get("anbieter") or "openrouter",
no_llm,
req.get("key") or None,
),
daemon=True,
)
t.start()
return {"job_id": job_id}
@app.get("/api/lauf/{job_id}")
def api_status(job_id: str) -> dict:
job = _jobs.get(job_id)
if not job:
raise HTTPException(status_code=404, detail="unbekannter Job")
return job
def _run_job(job_id, plz, modell, anbieter, no_llm, key) -> None:
job = _jobs[job_id]
try:
from .fetch import hole_angebote
from .modell import KategorisiertesAngebot
from .uebersicht import als_struktur
fetch = hole_angebote(plz) # deterministisch; AbbruchFehler bei Regel 4
if no_llm:
# Ohne LLM: belegte Rohliste, sichtbar als unkategorisiert markiert.
from .config import FALLBACK_GRUPPE
kat = [
KategorisiertesAngebot(a, FALLBACK_GRUPPE, unsicher=True)
for a in fetch.angebote
]
else:
from .kategorisieren import baue_kategorisierer, kategorisiere
job["phase"] = "kategorisieren"
kt = baue_kategorisierer(anbieter, modell, api_key=key)
def fort(done, total):
job["done"] = done
job["total"] = total
kat = kategorisiere(list(fetch.angebote), kt, fortschritt=fort)
job["ergebnis"] = als_struktur(fetch, kat)
job["status"] = "fertig"
_ergebnis_cache[(plz, modell, no_llm)] = job["ergebnis"]
except AbbruchFehler as e:
job["status"] = "fehler"
job["fehler"] = e.als_text()
except Exception as e: # nichts verstecken -- ehrliche Fehlermeldung
job["status"] = "fehler"
job["fehler"] = f"Unerwarteter Fehler: {e}"

View file

@ -0,0 +1,310 @@
<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>Angebots-Übersicht</title>
<style>
:root {
--bg:#f6f5f2; --panel:#ffffff; --ink:#1c1b19; --muted:#6b6864;
--line:#e2ded7; --accent:#1f6f53; --accent-weak:#e7f1ec;
--warn:#9a6a00; --warn-bg:#fdf3df; --price:#11221b;
--radius:10px; --mono:ui-monospace,"SF Mono",Menlo,Consolas,monospace;
}
* { box-sizing:border-box; }
body {
margin:0; background:var(--bg); color:var(--ink);
font:15px/1.5 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,sans-serif;
}
header.top {
padding:22px 26px 16px; border-bottom:1px solid var(--line); background:var(--panel);
}
header.top h1 { margin:0; font-size:20px; letter-spacing:-.01em; }
header.top p { margin:4px 0 0; color:var(--muted); font-size:13px; max-width:60ch; }
.wrap { max-width:1040px; margin:0 auto; padding:20px 26px 80px; }
.controls {
display:flex; flex-wrap:wrap; gap:10px 14px; align-items:end;
background:var(--panel); border:1px solid var(--line); border-radius:var(--radius);
padding:16px; margin-bottom:18px;
}
.field { display:flex; flex-direction:column; gap:4px; }
.field label { font-size:11px; text-transform:uppercase; letter-spacing:.05em; color:var(--muted); }
input, select, button { font:inherit; }
input[type=text], select {
padding:8px 10px; border:1px solid var(--line); border-radius:7px; background:#fff; color:var(--ink);
}
input#plz { width:90px; font-variant-numeric:tabular-nums; }
select#modell { min-width:230px; }
input#modellsuche { width:130px; }
.check { flex-direction:row; align-items:center; gap:7px; }
.check label { text-transform:none; letter-spacing:0; font-size:13px; color:var(--ink); }
button {
padding:9px 16px; border:1px solid var(--accent); border-radius:7px;
background:var(--accent); color:#fff; cursor:pointer; font-weight:600;
}
button.ghost { background:#fff; color:var(--accent); }
button:disabled { opacity:.5; cursor:default; }
button.icon { padding:8px 11px; }
.keyrow { width:100%; display:none; gap:8px; align-items:center; margin-top:4px; }
.keyrow.show { display:flex; }
.keyrow input { flex:1; }
.keytoggle { background:none; border:none; color:var(--muted); cursor:pointer; font-size:12px; text-decoration:underline; padding:0; }
#status { margin:14px 0; min-height:22px; font-size:14px; }
.bar { height:8px; background:var(--line); border-radius:5px; overflow:hidden; margin-top:8px; }
.bar > i { display:block; height:100%; width:0; background:var(--accent); transition:width .25s; }
.err { background:var(--warn-bg); border:1px solid #eccf8a; color:#5a3d00; padding:12px 14px; border-radius:8px; white-space:pre-wrap; font-family:var(--mono); font-size:13px; }
.summary { display:flex; flex-wrap:wrap; gap:10px 18px; align-items:center; color:var(--muted); font-size:13px; margin:6px 0 14px; }
.summary b { color:var(--ink); }
.filters { display:flex; flex-wrap:wrap; gap:10px 14px; align-items:end; margin-bottom:18px; }
section.grp { background:var(--panel); border:1px solid var(--line); border-radius:var(--radius); margin-bottom:12px; overflow:hidden; }
section.grp > h2 {
margin:0; padding:11px 16px; font-size:14px; display:flex; justify-content:space-between;
align-items:center; cursor:pointer; background:linear-gradient(#fff,#fbfaf7); border-bottom:1px solid var(--line);
}
section.grp > h2 .cnt { color:var(--muted); font-weight:500; font-variant-numeric:tabular-nums; }
section.grp.zu ul { display:none; }
section.grp ul { list-style:none; margin:0; padding:4px 0; }
li.ang { display:grid; grid-template-columns:1fr auto; gap:2px 14px; padding:9px 16px; border-bottom:1px solid #f1edea; align-items:baseline; }
li.ang:last-child { border-bottom:none; }
.ang .name { font-weight:600; }
.ang .name .marke { color:var(--accent); }
.ang .meta { grid-column:1; color:var(--muted); font-size:12.5px; }
.ang .meta .haendler { color:var(--ink); background:var(--accent-weak); padding:1px 6px; border-radius:5px; }
.ang .preis { grid-column:2; grid-row:1; text-align:right; font-weight:700; color:var(--price); font-variant-numeric:tabular-nums; white-space:nowrap; }
.ang .grund { grid-column:2; grid-row:2; text-align:right; color:var(--muted); font-size:12px; white-space:nowrap; }
.ang.unsicher { background:var(--warn-bg); }
.badge { display:inline-block; font-size:11px; background:var(--warn-bg); color:var(--warn); border:1px solid #eccf8a; border-radius:5px; padding:0 6px; margin-left:6px; }
.leer { padding:12px 16px; color:var(--muted); font-style:italic; }
.quelle { font-family:var(--mono); font-size:11px; color:#9a968f; }
footer.note { margin-top:22px; padding-top:16px; border-top:1px solid var(--line); color:var(--muted); font-size:12.5px; }
footer.note .haendler { color:var(--ink); }
footer.note .hinweis { display:block; margin-top:6px; font-style:italic; }
</style>
</head>
<body>
<header class="top">
<h1>Angebots-Übersicht</h1>
<p>Ortskonkret, händlerübergreifend, nach Produktgruppen. Daten deterministisch aus marktguru,
Einordnung per LLM. Jedes Angebot ist belegt — kein Auffüllen, Unsicheres ist markiert.</p>
</header>
<div class="wrap">
<div class="controls">
<div class="field"><label for="plz">PLZ</label><input id="plz" type="text" value="60487" inputmode="numeric" /></div>
<div class="field">
<label for="modell">Modell</label>
<select id="modell"><option>lädt…</option></select>
</div>
<div class="field"><label for="modellsuche">Modell suchen</label><input id="modellsuche" type="text" placeholder="z.B. qwen" /></div>
<div class="field"><label>&nbsp;</label><button id="refresh" class="ghost icon" title="Modell-Liste aktualisieren"></button></div>
<div class="field check"><input id="nollm" type="checkbox" /><label for="nollm">ohne LLM (Rohliste)</label></div>
<div class="field"><label>&nbsp;</label><button id="holen">Holen ▶</button></div>
<button type="button" class="keytoggle" id="keytoggle">API-Key setzen (optional)</button>
<div class="keyrow" id="keyrow">
<input id="key" type="password" placeholder="OPENROUTER_API_KEY (bleibt lokal, nur an deinen Server)" />
</div>
</div>
<div id="status"></div>
<div id="result" hidden>
<div class="summary" id="summary"></div>
<div class="filters">
<div class="field"><label for="fhaendler">Händler</label><select id="fhaendler"><option value="">alle</option></select></div>
<div class="field"><label for="ftext">Angebot suchen</label><input id="ftext" type="text" placeholder="Titel…" /></div>
<div class="field check"><input id="fsicher" type="checkbox" /><label for="fsicher">nur sichere</label></div>
</div>
<div id="gruppen"></div>
<footer class="note" id="footer"></footer>
</div>
</div>
<script>
const $ = s => document.querySelector(s);
let DATEN = null;
async function ladeModelle(q="") {
const sel = $("#modell");
try {
const r = await fetch("/api/modelle?q=" + encodeURIComponent(q));
if (!r.ok) throw new Error((await r.json()).detail || r.status);
const liste = await r.json();
sel.innerHTML = "";
if (!liste.length) { sel.innerHTML = "<option value=''>(keine Treffer)</option>"; return; }
for (const m of liste) {
const o = document.createElement("option");
o.value = m.id;
const tag = m.frei ? "FREE" : "paid";
const tools = m.tools ? "" : " ⚠ kein tool-calling";
o.textContent = `${m.id} [${tag}]${tools}`;
if (!m.tools) o.disabled = true;
sel.appendChild(o);
}
} catch (e) {
sel.innerHTML = `<option value=''>Modelle nicht abrufbar</option>`;
setStatus(`<div class="err">Modelle nicht abrufbar: ${e.message}</div>`);
}
}
function setStatus(html) { $("#status").innerHTML = html; }
async function starteLauf() {
const plz = $("#plz").value.trim();
if (!plz) { setStatus(`<div class="err">Bitte eine PLZ eingeben.</div>`); return; }
$("#holen").disabled = true;
$("#result").hidden = true;
const noLlm = $("#nollm").checked;
setStatus(`<div>Hole Angebote für <b>${plz}</b></div>`);
let job;
try {
const r = await fetch("/api/lauf", {
method:"POST", headers:{"Content-Type":"application/json"},
body: JSON.stringify({
plz, modell: $("#modell").value, no_llm: noLlm,
anbieter: "openrouter", key: $("#key").value || undefined
})
});
if (!r.ok) throw new Error((await r.json()).detail || r.status);
job = (await r.json()).job_id;
} catch (e) {
setStatus(`<div class="err">${e.message}</div>`); $("#holen").disabled = false; return;
}
const poll = setInterval(async () => {
let s;
try { s = await (await fetch("/api/lauf/" + job)).json(); }
catch { return; }
if (s.status === "laufend") {
if (s.phase === "kategorisieren" && s.total) {
const pct = Math.round(100 * s.done / s.total);
setStatus(`<div>Kategorisiere … Batch <b>${s.done}/${s.total}</b>
<div class="bar"><i style="width:${pct}%"></i></div></div>`);
} else {
setStatus(`<div>${s.phase === "kategorisieren" ? "Kategorisiere …" : "Hole Angebote …"}</div>`);
}
return;
}
clearInterval(poll);
$("#holen").disabled = false;
if (s.status === "fehler") { setStatus(`<div class="err">${s.fehler}</div>`); return; }
setStatus("");
DATEN = s.ergebnis;
render();
}, 1000);
}
function render() {
const d = DATEN;
$("#result").hidden = false;
$("#summary").innerHTML =
`Ort <b>${d.ort_name || d.ort_plz}</b> (PLZ ${d.ort_plz}) · <b>${d.anzahl}</b> Angebote ·
Quellen: ${d.quellen.join(", ") || "—"} · <b>${d.unsicher}</b> unsicher`;
const fh = $("#fhaendler");
const aktuell = fh.value;
fh.innerHTML = "<option value=''>alle</option>" +
d.haendler.map(h => `<option>${esc(h)}</option>`).join("");
fh.value = aktuell;
zeichneGruppen();
$("#footer").innerHTML =
`<span class="haendler"><b>Beobachtete Händler (belegt):</b> ${d.haendler.map(esc).join(", ")}.</span>` +
d.hinweise.map(h => `<span class="hinweis">${esc(h)}</span>`).join("");
}
function zeichneGruppen() {
const d = DATEN;
const fH = $("#fhaendler").value;
const fT = $("#ftext").value.trim().toLowerCase();
const nurSicher = $("#fsicher").checked;
const box = $("#gruppen");
box.innerHTML = "";
for (const g of d.gruppen) {
const items = g.angebote.filter(a =>
(!fH || a.haendler === fH) &&
(!nurSicher || !a.unsicher) &&
(!fT || (a.titel + " " + (a.marke||"")).toLowerCase().includes(fT))
);
const sec = document.createElement("section");
sec.className = "grp";
const h = document.createElement("h2");
h.innerHTML = `<span>${esc(g.name)}</span><span class="cnt">${items.length}</span>`;
h.onclick = () => sec.classList.toggle("zu");
sec.appendChild(h);
if (!items.length) {
const p = document.createElement("div"); p.className = "leer";
p.textContent = (fH || fT || nurSicher) ? "keine Treffer im Filter" : "keine Angebote";
sec.appendChild(p);
} else {
const ul = document.createElement("ul");
for (const a of items) ul.appendChild(zeile(a));
sec.appendChild(ul);
}
box.appendChild(sec);
}
}
function zeile(a) {
const li = document.createElement("li");
li.className = "ang" + (a.unsicher ? " unsicher" : "");
const marke = a.marke ? `<span class="marke">${esc(a.marke)}</span> ` : "";
const menge = a.menge ? ` · ${esc(a.menge)}` : "";
const gueltig = (a.gueltig_von || a.gueltig_bis)
? ` · gültig ${a.gueltig_von||"?"}${a.gueltig_bis||"?"}` : "";
const preis = a.preis != null
? a.preis.toLocaleString("de-DE",{minimumFractionDigits:2,maximumFractionDigits:2}) + " €"
: "Preis fehlt";
const badge = a.unsicher ? `<span class="badge">unsicher</span>` : "";
li.innerHTML =
`<div class="name">${marke}${esc(a.titel)}${badge}</div>` +
`<div class="preis">${preis}</div>` +
`<div class="meta"><span class="haendler">${esc(a.haendler)}</span>${menge}${gueltig}
&nbsp;<span class="quelle" title="${esc(a.quelle)}">${esc((a.quelle||"").slice(0,42))}</span></div>` +
`<div class="grund">${a.grundpreis ? esc(a.grundpreis) : ""}</div>`;
return li;
}
function esc(s){ return (s==null?"":String(s)).replace(/[&<>"]/g,c=>({"&":"&amp;","<":"&lt;",">":"&gt;",'"':"&quot;"}[c])); }
// Events
$("#holen").onclick = starteLauf;
$("#refresh").onclick = () => ladeModelle($("#modellsuche").value.trim());
$("#modellsuche").addEventListener("keydown", e => { if (e.key==="Enter") ladeModelle(e.target.value.trim()); });
$("#keytoggle").onclick = () => $("#keyrow").classList.toggle("show");
for (const id of ["#fhaendler","#ftext","#fsicher"])
$(id).addEventListener("input", () => DATEN && zeichneGruppen());
$("#plz").addEventListener("keydown", e => { if (e.key==="Enter") starteLauf(); });
// Auto-Start per URL-Param: ?plz=60487[&no_llm=1][&modell=...]
const _p = new URLSearchParams(location.search);
if (_p.get("plz")) {
$("#plz").value = _p.get("plz");
if (_p.get("no_llm")) $("#nollm").checked = true;
ladeModelle().then(() => {
const m = _p.get("modell");
if (m) {
const sel = $("#modell");
if (![...sel.options].some(o => o.value === m)) {
const o = document.createElement("option");
o.value = o.textContent = m;
sel.insertBefore(o, sel.firstChild);
}
sel.value = m;
}
starteLauf();
});
} else {
ladeModelle();
}
</script>
</body>
</html>

0
tests/__init__.py Normal file
View file

54
tests/fakes.py Normal file
View file

@ -0,0 +1,54 @@
"""Test-Doubles -- halten die Tests offline (kein Netz, kein LLM, kein Schlüssel)."""
from __future__ import annotations
from datetime import date, datetime
from typing import Callable
from angebote.modell import Angebot
from angebote.quellen.basis import Ort
class FakeQuelle:
"""Adapter-Double: liefert vorgegebene Angebote, deckt-Flag, optional Fehler."""
def __init__(self, name, angebote=None, *, deckt=True, fehler=None):
self.name = name
self._angebote = list(angebote or [])
self._deckt = deckt
self._fehler = fehler
def deckt_ab(self, ort: Ort) -> bool:
return self._deckt
def hole(self, ort: Ort):
if self._fehler is not None:
raise self._fehler
return list(self._angebote)
class FakeKategorisierer:
"""Kategorisierer-Double: ruft eine reine Funktion posten->antworten auf."""
def __init__(self, fn: Callable[[list[dict]], list[dict]]):
self._fn = fn
def klassifiziere(self, posten: list[dict]) -> list[dict]:
return self._fn(posten)
def beispiel_angebot(titel="Butter", **kw) -> Angebot:
"""Belegtes Angebot mit Default-Pflichtfeldern; einzeln überschreibbar."""
daten = dict(
titel=titel,
haendler="REWE",
quelle="test:fixture",
abgerufen_am=datetime(2026, 6, 1, 8, 0, 0),
preis=1.49,
marke="Markenbutter",
menge="250 g",
gueltig_von=date(2026, 6, 1),
gueltig_bis=date(2026, 6, 7),
)
daten.update(kw)
return Angebot(**daten)

View file

@ -0,0 +1,41 @@
"""Regel 4: Abbruch statt stiller Drift -- mit Schwelle, Ursache, Vorschlag."""
import pytest
from angebote.fehler import AbbruchFehler
from angebote.fetch import aufloesen_ort, hole_angebote
from tests.fakes import FakeQuelle, beispiel_angebot
def test_unaufloesbarer_ort_bricht_ab():
with pytest.raises(AbbruchFehler) as exc:
aufloesen_ort("Hintertupfingen")
e = exc.value
# Der Abbruch ist brauchbar: alle drei Felder sind belegt.
assert e.schwelle and e.ursache and e.vorschlag
assert "PLZ" in e.vorschlag
def test_plz_wird_direkt_aufgeloest():
ort = aufloesen_ort("60487")
assert ort.plz == "60487"
def test_bekannter_ortsname_wird_aufgeloest():
ort = aufloesen_ort("Frankfurt")
assert ort.plz and ort.plz.isdigit()
def test_keine_quelle_deckt_ort_ab_bricht_ab():
with pytest.raises(AbbruchFehler) as exc:
hole_angebote("60487", [FakeQuelle("woanders", [], deckt=False)])
assert exc.value.schwelle == "Ortsabdeckung"
def test_adapter_abbruch_propagiert_nicht_kaschiert():
# Eine Quelle, die ihren Ortsbezug nicht herstellen kann, bricht ab --
# der Orchestrator schluckt das NICHT zu einem leeren Ergebnis.
fehler = AbbruchFehler("Quelle X", "Ortsbezug nicht verifiziert", "andere Quelle")
quelle = FakeQuelle("kaputt", [beispiel_angebot()], fehler=fehler)
with pytest.raises(AbbruchFehler):
hole_angebote("60487", [quelle])

View file

@ -0,0 +1,21 @@
"""Regel: Kein Auffüllen. Leeres Quellergebnis bleibt leer."""
from angebote.fetch import hole_angebote
from angebote.uebersicht import rendern
from tests.fakes import FakeQuelle
def test_leere_quelle_fuellt_nicht_auf():
ergebnis = hole_angebote("60487", [FakeQuelle("leer", [])])
assert ergebnis.angebote == ()
# Der Lauf wird ehrlich vermerkt, nicht versteckt:
assert any("0 Angebote" in h for h in ergebnis.hinweise)
def test_uebersicht_zeigt_keine_erfundenen_beispiele():
ergebnis = hole_angebote("60487", [FakeQuelle("leer", [])])
text = rendern(ergebnis, [])
# Jede Gruppe steht da -- aber leer, als "keine Angebote", nicht aufgefüllt.
assert text.count("_keine Angebote_") >= 1
# Keine Beispieldaten:
assert "" not in text or "Preis fehlt" in text

View file

@ -0,0 +1,35 @@
"""Regel: Geschlossene Kategorienliste. Off-list -> Fallback + unsicher."""
from angebote.config import FALLBACK_GRUPPE, PRODUKTGRUPPEN
from angebote.kategorisieren import kategorisiere
from tests.fakes import FakeKategorisierer, beispiel_angebot
def test_erfundene_gruppe_wird_nicht_uebernommen():
# Das Modell "erfindet" eine Gruppe -> Code zwingt sie in den Fallback.
fake = FakeKategorisierer(
lambda posten: [
{"id": p["id"], "gruppe": "Weltraumzeug", "unsicher": False} for p in posten
]
)
ergebnis = kategorisiere([beispiel_angebot("Mondstaub")], fake)
assert ergebnis[0].gruppe == FALLBACK_GRUPPE
assert ergebnis[0].unsicher is True
def test_keine_gruppe_ausserhalb_der_config():
namen = ["Butter", "Apfel", "Cola", "Bier", "Pizza", "Shampoo", "Bohrmaschine"]
fake = FakeKategorisierer(
# Mischung aus gültigen und ungültigen Gruppen:
lambda posten: [
{
"id": p["id"],
"gruppe": "Obst & Gemüse" if i % 2 == 0 else "Quatschgruppe",
"unsicher": False,
}
for i, p in enumerate(posten)
]
)
ergebnis = kategorisiere([beispiel_angebot(n) for n in namen], fake)
for ka in ergebnis:
assert ka.gruppe in PRODUKTGRUPPEN

View file

@ -0,0 +1,53 @@
"""Regel: Daten sind unantastbar. Preis/Gültigkeit/Händler bleiben identisch."""
from dataclasses import replace
from angebote.kategorisieren import kategorisiere
from tests.fakes import FakeKategorisierer, beispiel_angebot
def _gib_gruppe(gruppe):
return FakeKategorisierer(
lambda posten: [{"id": p["id"], "gruppe": gruppe, "unsicher": False} for p in posten]
)
def test_originaldaten_unveraendert_property():
angebote = [
beispiel_angebot("Butter", preis=1.49, haendler="REWE"),
beispiel_angebot("Hähnchen", preis=None, haendler="Penny", marke=None),
beispiel_angebot("Wein", preis=4.99, haendler="EDEKA", menge="0,75 l"),
]
ergebnis = kategorisiere(angebote, _gib_gruppe("Sonstiges"))
nach = {k.angebot.angebot_id: k.angebot for k in ergebnis}
for vorher in angebote:
nachher = nach[vorher.angebot_id]
# Property über den ganzen Stream: jedes belegte Feld bleibt gleich.
assert nachher.titel == vorher.titel
assert nachher.preis == vorher.preis
assert nachher.haendler == vorher.haendler
assert nachher.marke == vorher.marke
assert nachher.menge == vorher.menge
assert nachher.gueltig_von == vorher.gueltig_von
assert nachher.gueltig_bis == vorher.gueltig_bis
# Sogar Objekt-Identität: das eingefrorene Original wird durchgereicht.
assert nachher is vorher
def test_fehlender_preis_bleibt_fehlend():
angebot = beispiel_angebot("Hähnchen", preis=None)
ergebnis = kategorisiere([angebot], _gib_gruppe("Fleisch & Wurst"))
assert ergebnis[0].angebot.preis is None
def test_modell_kann_daten_strukturell_nicht_aendern():
# frozen=True: ein Versuch, den Preis zu "korrigieren", schlägt fehl.
angebot = beispiel_angebot("Butter", preis=1.49)
import pytest
with pytest.raises(Exception):
angebot.preis = 0.99 # type: ignore[misc]
# replace erzeugt ein NEUES Objekt -- das Original bleibt unberührt.
anderes = replace(angebot, preis=0.99)
assert angebot.preis == 1.49 and anderes.preis == 0.99

View file

@ -0,0 +1,41 @@
"""Regel: Unsicherheit wird geflaggt, nicht still einsortiert."""
from angebote.kategorisieren import kategorisiere
from angebote.modell import FetchErgebnis
from angebote.uebersicht import rendern
from tests.fakes import FakeKategorisierer, beispiel_angebot
def test_mehrdeutiger_artikel_wird_geflaggt():
fake = FakeKategorisierer(
lambda posten: [
{"id": p["id"], "gruppe": "Molkereiprodukte & Eier", "unsicher": True}
for p in posten
]
)
ergebnis = kategorisiere([beispiel_angebot("Hafer-Pflanzendrink")], fake)
assert ergebnis[0].unsicher is True
def test_nicht_beantworteter_posten_wird_unsicher_statt_still():
# Modell antwortet zu KEINEM Posten -> kein stilles Einsortieren.
fake = FakeKategorisierer(lambda posten: [])
ergebnis = kategorisiere([beispiel_angebot("Rätselartikel")], fake)
assert ergebnis[0].unsicher is True
def test_unsicherheit_ist_in_der_uebersicht_sichtbar():
fake = FakeKategorisierer(
lambda posten: [
{"id": p["id"], "gruppe": "Sonstiges", "unsicher": True} for p in posten
]
)
kat = kategorisiere([beispiel_angebot("Pflanzendrink")], fake)
fetch = FetchErgebnis(
ort_plz="60487",
ort_name=None,
angebote=tuple(k.angebot for k in kat),
abgedeckte_quellen=("test",),
)
text = rendern(fetch, kat)
assert "unsicher" in text

View file

@ -0,0 +1,144 @@
"""Anbieter-Factory + OpenRouter-Antwortform -- offline, ohne Netz/Key."""
import json
import pytest
from angebote.fehler import AbbruchFehler
from angebote.kategorisieren import (
OpenRouterKategorisierer,
baue_kategorisierer,
kategorisiere,
)
from tests.fakes import beispiel_angebot
class _FakeResp:
def __init__(self, payload, status_code=200):
self._p = payload
self.status_code = status_code
self.headers = {}
def raise_for_status(self):
pass
def json(self):
return self._p
class _FakeSession:
def __init__(self, payload):
self._p = payload
self.calls = []
def post(self, url, json=None, headers=None, timeout=None):
self.calls.append({"url": url, "json": json, "headers": headers})
return _FakeResp(self._p)
def _openrouter_payload(zuordnungen):
return {
"choices": [
{
"message": {
"tool_calls": [
{
"function": {
"name": "zuordnungen",
"arguments": json.dumps({"zuordnungen": zuordnungen}),
}
}
]
}
}
]
}
def test_factory_unbekannter_anbieter_bricht_ab():
with pytest.raises(AbbruchFehler):
baue_kategorisierer("gibtsnicht")
def test_factory_openrouter_ohne_key_bricht_ehrlich_ab(monkeypatch):
monkeypatch.delenv("OPENROUTER_API_KEY", raising=False)
with pytest.raises(AbbruchFehler) as exc:
baue_kategorisierer("openrouter")
assert "OPENROUTER_API_KEY" in exc.value.ursache
def test_openrouter_parst_openai_tool_calls():
angebote = [beispiel_angebot("Apfel"), beispiel_angebot("Bohrmaschine")]
payload = _openrouter_payload(
[
{"id": angebote[0].angebot_id, "gruppe": "Obst & Gemüse", "unsicher": False},
{
"id": angebote[1].angebot_id,
"gruppe": "Non-Food (Kleidung, Spielzeug, Garten, Technik)",
"unsicher": False,
},
]
)
sess = _FakeSession(payload)
kat = OpenRouterKategorisierer(api_key="test-key", session=sess)
ergebnis = kategorisiere(angebote, kat)
gruppen = {k.angebot.titel: k.gruppe for k in ergebnis}
assert gruppen["Apfel"] == "Obst & Gemüse"
assert gruppen["Bohrmaschine"].startswith("Non-Food")
# Request war OpenAI-kompatibel adressiert und authentifiziert:
aufruf = sess.calls[0]
assert aufruf["url"].endswith("/chat/completions")
assert aufruf["json"]["tool_choice"]["function"]["name"] == "zuordnungen"
assert aufruf["headers"]["Authorization"] == "Bearer test-key"
def test_openrouter_api_fehler_bricht_ab():
sess = _FakeSession({"error": {"message": "insufficient_credits"}})
kat = OpenRouterKategorisierer(api_key="test-key", session=sess)
with pytest.raises(AbbruchFehler):
kat.klassifiziere([{"id": "x", "titel": "Apfel"}])
class _SeqSession:
"""Liefert eine Folge vorgegebener Antworten (für Retry-Tests)."""
def __init__(self, responses):
self._responses = list(responses)
self.calls = 0
def post(self, url, json=None, headers=None, timeout=None):
r = self._responses[min(self.calls, len(self._responses) - 1)]
self.calls += 1
return r
def test_openrouter_retry_bei_429_dann_erfolg():
payload = _openrouter_payload(
[{"id": "x", "gruppe": "Sonstiges", "unsicher": False}]
)
seq = _SeqSession(
[_FakeResp({}, 429), _FakeResp({}, 429), _FakeResp(payload, 200)]
)
geschlafen = []
kat = OpenRouterKategorisierer(
api_key="k",
session=seq,
mindest_abstand_s=0.0,
schlafen=geschlafen.append,
)
res = kat.klassifiziere([{"id": "x", "titel": "Apfel"}])
assert res == [{"id": "x", "gruppe": "Sonstiges", "unsicher": False}]
assert seq.calls == 3
assert len(geschlafen) == 2 # zwei Backoffs vor dem Erfolg, kein echtes Warten
def test_openrouter_429_erschoepft_bricht_als_ratelimit_ab():
seq = _SeqSession([_FakeResp({}, 429)])
kat = OpenRouterKategorisierer(
api_key="k", session=seq, max_versuche=3, mindest_abstand_s=0.0, schlafen=lambda s: None
)
with pytest.raises(AbbruchFehler) as exc:
kat.klassifiziere([{"id": "x", "titel": "Apfel"}])
assert "Rate-Limit" in exc.value.schwelle
assert seq.calls == 3

136
tests/test_modelle.py Normal file
View file

@ -0,0 +1,136 @@
"""Modell-Discovery + interaktive Auswahl -- offline, ohne Netz/Key."""
from angebote.modelle import lade_modelle, parse_modelle, suche, top_free
from angebote.modellauswahl import waehle_modell_interaktiv
FAKE = [
{
"id": "moonshotai/kimi-k2.6:free",
"name": "Kimi K2",
"context_length": 262144,
"pricing": {"prompt": "0", "completion": "0"},
"supported_parameters": ["tools", "response_format"],
},
{
"id": "meta-llama/llama-3.3-70b-instruct:free",
"name": "Llama 3.3 70B",
"context_length": 131072,
"pricing": {"prompt": "0", "completion": "0"},
"supported_parameters": ["tools"],
},
{
"id": "some/free-no-tools:free",
"name": "Free ohne Tools",
"context_length": 100000,
"pricing": {"prompt": "0", "completion": "0"},
"supported_parameters": [],
},
{
"id": "anthropic/claude-sonnet-4.6",
"name": "Claude Sonnet 4.6",
"context_length": 200000,
"pricing": {"prompt": "3", "completion": "15"},
"supported_parameters": ["tools"],
},
]
class _Resp:
def __init__(self, payload):
self._p = payload
def raise_for_status(self):
pass
def json(self):
return self._p
class _Session:
def __init__(self, data):
self._data = data
def get(self, url, timeout=None):
return _Resp({"data": self._data})
def _scripted(antworten):
it = iter(antworten)
return lambda prompt="": next(it)
# --- reine Daten-Logik -------------------------------------------------------
def test_parse_erkennt_frei_und_tools():
modelle = parse_modelle(FAKE)
nach_id = {m.id: m for m in modelle}
assert nach_id["moonshotai/kimi-k2.6:free"].frei is True
assert nach_id["moonshotai/kimi-k2.6:free"].tools is True
assert nach_id["anthropic/claude-sonnet-4.6"].frei is False
assert nach_id["some/free-no-tools:free"].tools is False
def test_top_free_nur_tools_und_rangfolge():
modelle = parse_modelle(FAKE)
top = top_free(modelle, 5, nur_tools=True)
ids = [m.id for m in top]
# paid (Sonnet) und das tool-lose Free-Modell sind raus:
assert "anthropic/claude-sonnet-4.6" not in ids
assert "some/free-no-tools:free" not in ids
# Präferenz: kimi-k2 vor llama-3.3-70b
assert ids == [
"moonshotai/kimi-k2.6:free",
"meta-llama/llama-3.3-70b-instruct:free",
]
def test_suche_findet_teilstring():
modelle = parse_modelle(FAKE)
treffer = suche(modelle, "llama")
assert [m.id for m in treffer] == ["meta-llama/llama-3.3-70b-instruct:free"]
def test_lade_modelle_ueber_session():
modelle = lade_modelle(session=_Session(FAKE))
assert len(modelle) == 4
# --- interaktiver Picker -----------------------------------------------------
def test_picker_waehlt_per_nummer():
gewaehlt = waehle_modell_interaktiv(
session=_Session(FAKE),
eingabe=_scripted(["1"]),
ausgabe=lambda s: None,
)
assert gewaehlt == "moonshotai/kimi-k2.6:free"
def test_picker_suche_dann_wahl():
gewaehlt = waehle_modell_interaktiv(
session=_Session(FAKE),
eingabe=_scripted(["s llama", "1"]),
ausgabe=lambda s: None,
)
assert gewaehlt == "meta-llama/llama-3.3-70b-instruct:free"
def test_picker_warnt_bei_modell_ohne_tools_und_bricht_ab():
# Suche bringt das tool-lose Modell in die Liste; Wahl -> Warnung -> 'n' -> q.
gewaehlt = waehle_modell_interaktiv(
session=_Session(FAKE),
eingabe=_scripted(["s no-tools", "1", "n", "q"]),
ausgabe=lambda s: None,
)
assert gewaehlt is None
def test_picker_quit_gibt_none():
gewaehlt = waehle_modell_interaktiv(
session=_Session(FAKE),
eingabe=_scripted(["q"]),
ausgabe=lambda s: None,
)
assert gewaehlt is None

36
tests/test_schnitt.py Normal file
View file

@ -0,0 +1,36 @@
"""Der Schnitt als Test: im Fetch-Teil darf KEIN LLM geladen werden.
Geprüft in einem frischen Interpreter: nach dem Import des kompletten
Fetch-Pfads (fetch + marktguru-Adapter) darf 'anthropic' nicht in sys.modules
stehen. So bleibt die Trennung 'kein LLM im Fetch-Teil' eine prüfbare Bedingung.
"""
import subprocess
import sys
from pathlib import Path
SRC = Path(__file__).resolve().parents[1] / "src"
def test_fetch_teil_laedt_kein_anthropic():
code = (
"import sys; "
"import angebote.fetch; "
"import angebote.quellen.marktguru; "
"assert 'anthropic' not in sys.modules, "
"'Fetch-Teil hat anthropic geladen -- Schnitt verletzt'; "
"print('ok')"
)
proc = subprocess.run(
[sys.executable, "-c", code],
cwd=str(SRC),
capture_output=True,
text=True,
)
assert proc.returncode == 0, proc.stderr
assert "ok" in proc.stdout
def test_marktguru_quelltext_nennt_anthropic_nicht():
quelltext = (SRC / "angebote" / "quellen" / "marktguru.py").read_text("utf-8")
assert "anthropic" not in quelltext.lower()

78
tests/test_web.py Normal file
View file

@ -0,0 +1,78 @@
"""Web-Schicht -- offline. Struktur-Funktion + Endpoints (ohne Netz/Key/LLM).
Der Schnitt gilt auch hier: die Web-Schicht fügt keine Logik hinzu, sie ruft
die getesteten Module auf. Geprüft wird, dass sie das belegt-und-ehrlich
weiterreicht.
"""
import pytest
from angebote.modell import FetchErgebnis, KategorisiertesAngebot
from angebote.uebersicht import als_struktur
from tests.fakes import beispiel_angebot
fastapi = pytest.importorskip("fastapi")
from fastapi.testclient import TestClient # noqa: E402
import angebote.web as web # noqa: E402
from angebote.modelle import ModellInfo # noqa: E402
def _fetch_mit(angebote):
return FetchErgebnis(
ort_plz="60487",
ort_name=None,
angebote=tuple(angebote),
abgedeckte_quellen=("marktguru",),
gesehene_haendler=tuple(sorted({a.haendler for a in angebote})),
hinweise=("marktguru: Teilabdeckung",),
)
def test_als_struktur_behaelt_belegte_felder_und_leere_gruppen():
a = beispiel_angebot("Butter", preis=1.49, haendler="REWE")
kat = [KategorisiertesAngebot(a, "Molkereiprodukte & Eier", unsicher=False)]
struktur = als_struktur(_fetch_mit([a]), kat)
assert struktur["anzahl"] == 1
# alle 13 Produktgruppen vorhanden, leere als leere Liste (kein Weglassen):
assert len(struktur["gruppen"]) == 13
molk = next(g for g in struktur["gruppen"] if g["name"].startswith("Molkerei"))
assert molk["angebote"][0]["preis"] == 1.49
assert molk["angebote"][0]["haendler"] == "REWE"
leere = [g for g in struktur["gruppen"] if g["anzahl"] == 0]
assert leere # leere Gruppen bleiben erhalten
def test_index_liefert_html():
client = TestClient(web.app)
r = client.get("/")
assert r.status_code == 200
assert "Angebots-Übersicht" in r.text
def test_api_modelle_gibt_top_free(monkeypatch):
fake = [
ModellInfo("moonshotai/kimi-k2.6:free", "Kimi", 262144, True, True),
ModellInfo("anthropic/claude-sonnet-4.6", "Sonnet", 200000, False, True),
]
monkeypatch.setattr(web, "_jobs", {}) # sauberer Zustand
monkeypatch.setattr("angebote.modelle.lade_modelle", lambda session=None: fake)
client = TestClient(web.app)
r = client.get("/api/modelle")
assert r.status_code == 200
daten = r.json()
assert daten and daten[0]["id"] == "moonshotai/kimi-k2.6:free"
assert daten[0]["frei"] is True
def test_api_lauf_ohne_plz_ist_400():
client = TestClient(web.app)
r = client.post("/api/lauf", json={"plz": ""})
assert r.status_code == 400
def test_api_status_unbekannt_ist_404():
client = TestClient(web.app)
r = client.get("/api/lauf/gibtsnicht")
assert r.status_code == 404