"""PROTOTYP — wegwerfen. Lässt extraktion.py gegen die Testseiten laufen und schreibt ergebnis.json sowie bericht.html.""" import html import json import os import sys from concurrent.futures import ThreadPoolExecutor import requests from extraktion import als_jpeg_b64, extrahieren MODELL = "gpt-5.4-mini" KEY = os.environ.get("OPENAI_API_KEY", "") SEITEN = { "CER Container, Rostock": "https://cer-container.de/", "WUD Entsorgung, Saarland": "https://www.wud-entsorgung.de/", "Buhrfeind": "https://buhrfeind.de/", "Gieske Container": "https://containerdienst-gieske.de/", "Hildebrandt Container": "https://hildebrandt-containerdienst.de/", "Johann Schröder": "https://www.johann-schroeder.de/container/", } def frage(inhalt, schema): r = requests.post("https://api.openai.com/v1/chat/completions", timeout=180, headers={"Authorization": f"Bearer {KEY}"}, json={"model": MODELL, "messages": [{"role": "user", "content": inhalt}], "response_format": {"type": "json_schema", "json_schema": {"name": "antwort", "strict": True, "schema": schema}}}) r.raise_for_status() return json.loads(r.json()["choices"][0]["message"]["content"]) def bildteil(b64): return {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}", "detail": "low"}} CONTAINER_SCHEMA = {"type": "object", "additionalProperties": False, "required": ["bilder"], "properties": {"bilder": {"type": "array", "items": { "type": "object", "additionalProperties": False, "required": ["nr", "ist_container", "typ", "farbe_hex", "sicherheit", "begruendung"], "properties": { "nr": {"type": "integer"}, "ist_container": {"type": "boolean"}, "typ": {"type": "string", "enum": ["absetzcontainer", "abrollcontainer", "anderer", "keiner"]}, "farbe_hex": {"type": "string"}, "sicherheit": {"type": "number"}, "begruendung": {"type": "string"}, }}}}} CONTAINER_PROMPT = """Du prüfst Fotos von der Website eines deutschen Containerdienstes. Für jedes Bild (nummeriert in Reihenfolge ab 0): Zeigt es einen echten Entsorgungscontainer (Absetzcontainer/Mulde oder Abrollcontainer), dessen lackierte Außenwand gut sichtbar ist? Illustrationen, Icons, Grafiken, Mülltonnen, Seecontainer und Baucontainer zählen nicht. Wenn ja: gib die Lackfarbe der Containerwand als Hex an – so wie sie bei neutralem Tageslicht aussähe, nicht Schatten, Rost, Schmutz oder Ladung. Sicherheit 0 bis 1. Sonst farbe_hex leer lassen.""" def vision_container(fotos): urteile = [] for i in range(0, len(fotos), 8): paket = fotos[i:i + 8] inhalt = [{"type": "text", "text": CONTAINER_PROMPT}] + [bildteil(als_jpeg_b64(f["bild"])) for f in paket] for u in frage(inhalt, CONTAINER_SCHEMA)["bilder"]: if 0 <= u["nr"] < len(paket): urteile.append({**u, "url": paket[u["nr"]]["url"], "seite": paket[u["nr"]]["seite"]}) return urteile LOGO_SCHEMA = {"type": "object", "additionalProperties": False, "required": ["nr", "farbe_hex", "begruendung"], "properties": { "nr": {"type": "integer"}, "farbe_hex": {"type": "string"}, "begruendung": {"type": "string"}}} def vision_logo(kandidaten): from PIL import Image import io, subprocess, tempfile teile = [{"type": "text", "text": "Welches dieser Bilder (nummeriert ab 0) ist das Firmenlogo des Containerdienstes – das vollständige Logo, nicht ein Siegel, Partnerlogo oder Social-Media-Icon? Gib -1, wenn keins passt. farbe_hex: die markanteste Markenfarbe im Logo (nicht Schwarz/Weiß/Grau, außer es gibt keine andere)."}] brauchbar = [] for k in kandidaten: if k["svg"]: # SVG über macOS-QuickLook rastern with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "l.svg") open(p, "wb").write(k["bytes"]) subprocess.run(["qlmanage", "-t", "-s", "512", "-o", d, p], capture_output=True) png = p + ".png" if not os.path.exists(png): continue bild = Image.open(png); bild.load() else: bild = k["bild"] brauchbar.append(k) teile.append(bildteil(als_jpeg_b64(bild, 400))) if not brauchbar: return None a = frage(teile, LOGO_SCHEMA) if not 0 <= a["nr"] < len(brauchbar): return None return {"url": brauchbar[a["nr"]]["url"], "farbe_hex": a["farbe_hex"] or None, "begruendung": a["begruendung"]} def lauf(item): name, url = item print(f"{name}: {url}", flush=True) try: return name, extrahieren(url, vision_container, vision_logo, protokoll=lambda m: print(f"[{name}] {m}", flush=True)) except Exception as e: return name, {"website": url, "fehler": repr(e)} def bericht(ergebnisse): e = html.escape karten = [] for name, r in ergebnisse.items(): if "fehler" in r: karten.append(f"

{e(name)}

{e(r['fehler'])}

") continue logo = r["logo"] treffer = "".join( f"
" f"
{e(t['farbe_hex'])} · {e(t['typ'])} · {t['sicherheit']:.2f}
{e(t['begruendung'])}
" for t in r["container_treffer"]) or "

Kein Containerfoto erkannt.

" import re from bilder import ANSICHTEN slug = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-") hier = os.path.dirname(os.path.abspath(__file__)) def lade(datei): p = os.path.join(hier, datei) return (json.load(open(p)) if os.path.exists(p) else {}).get(slug, {}) varianten = [("gen", "1 · KI: Farbe und Logo", lade("bilder.json")), ("gen3", "3 · Code, Logo flach", lade("bilder3.json")), ("gen3s", "3 · Code, Aufkleber mit Struktur", lade("bilder-gen3s.json"))] erzeugt = varianten[0][2] def zelle(ordner, titel, daten, d): eintrag = daten.get(d) ok = isinstance(eintrag, str) and not eintrag.startswith("Fehler") or isinstance(eintrag, dict) and eintrag.get("pfad") inhalt = f"" if ok else "

–

" return f"
{inhalt}
{e(titel)}
" if erzeugt: zeilen = "".join( f"

{e(label)} · {'Farbe + Logo' if mit_logo else 'nur Farbe'}

" f"
EV-Vorlage
" + "".join(zelle(o, t, dd, d) for o, t, dd in varianten) + "
" for d, label, mit_logo in ANSICHTEN) generiert = f"

Erzeugte Containerbilder

{zeilen}" stufen_ordner = os.path.join(hier, "gen3k", slug) if os.path.isdir(stufen_ordner): from stufen import ANSICHTEN as ST_ANSICHTEN, STUFEN reihen = "".join( f"
Variante 1 (KI)
" + "".join(f"
Code · Kontrast ×{k:.1f}" + (f", Wand −{ab}" if ab else "") + "
" for kurz, k, ab in STUFEN) + "
" for d in ST_ANSICHTEN) generiert += f"

Kontraststufen (Variante 3) neben Variante 1

{reihen}" elif r.get("farbe_quelle") != "container": generiert = "

Erzeugte Containerbilder

Keine erzeugt: Ohne Containerfoto muss die Farbe erst ein Mensch prüfen.

" else: generiert = "" verworfen = "".join(f"
  • {e(t['url'].rsplit('/',1)[-1][:50])} – {t['sicherheit']:.2f}, {e(t['begruendung'])}
    • " for t in r["container_verworfen"]) karten.append(f"""

    {e(name)} {e(r['website'])}

    Logo

    {f"

    {e(logo['begruendung'])}

    " if logo else "

    Kein Logo gefunden.

    "}

    Containerfarbe

    {f"

    {e(r['farbe'])} aus {e(r['farbe_quelle'])}" + (f", {int(r['farbe_anteil']*100)} % der Treffer" if r['farbe_anteil'] else "") + "

    " if r['farbe'] else "

    Keine Farbe.

    "}

    Durchsucht

    {r['seiten']} Seiten · {r['bilder']} Bilder · {r['fotos_geprueft']} Fotos geprüft · {len(r['container_treffer'])} Containertreffer

    {generiert}

    Containerfotos, aus denen die Farbe stammt

    {treffer}
    {f"
    {len(r['container_verworfen'])} unsichere Containertreffer
    " if verworfen else ""}
    """) return f""" Logo und Containerfarbe

    Logo und Containerfarbe

    Prototyp: Das Skript durchsucht jede Website, wählt das Logo und leitet die Containerfarbe aus den erkannten Containerfotos ab. Findet es kein Containerfoto, nimmt es die Markenfarbe des Logos.

    {''.join(karten)}
    """ if __name__ == "__main__": auswahl = {k: v for k, v in SEITEN.items() if len(sys.argv) < 2 or any(a.lower() in k.lower() for a in sys.argv[1:])} with ThreadPoolExecutor(6) as pool: ergebnisse = dict(pool.map(lauf, auswahl.items())) ergebnisse = {k: ergebnisse[k] for k in auswahl} hier = os.path.dirname(os.path.abspath(__file__)) json.dump(ergebnisse, open(os.path.join(hier, "ergebnis.json"), "w"), ensure_ascii=False, indent=2) open(os.path.join(hier, "bericht.html"), "w").write(bericht(ergebnisse)) for name, r in ergebnisse.items(): print(f"{name}: Logo={(r.get('logo') or {}).get('url')} Farbe={r.get('farbe')} ({r.get('farbe_quelle')}) Treffer={len(r.get('container_treffer', []))} {r.get('fehler','')}")