Endlich Barrierefrei: ALT-Texte mit AI effizient füllen

1. Produkte und Bilder sammeln

Ziel
Alle Produktbilder aus Shopify abrufen, sauber strukturieren und lokal speichern.

Wie

• Das Skript holt Produkte über iter_products, extrahiert Bilder und baut dafür ImageRef Objekte.
• Relevante Felder: product_id, image_id, src, alt, product_title, vendor, variants.
• Nicht unterstützte Formate werden mit is_supported_image direkt aussortiert, etwa svg.

Datenbank

• db_init erzeugt die Tabelle jobs.
• Beim Einsammeln werden Einträge erst beim Verarbeiten geschrieben, dadurch bleibt die DB schlank.
• Du kannst optional eine reine Inventartabelle hinzufügen, wenn Du einen vollständigen Snapshot möchtest:

CREATE TABLE IF NOT EXISTS inventory(
  product_id INTEGER,
  image_id   INTEGER,
  src        TEXT,
  alt_shop   TEXT,
  title      TEXT,
  vendor     TEXT,
  ts         TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY(product_id, image_id)
);

Zum Füllen, direkt nach collect_images():

for ref in all_refs:
    conn.execute("""
      INSERT OR REPLACE INTO inventory(product_id,image_id,src,alt_shop,title,vendor)
      VALUES(?,?,?,?,?,?)
    """,(ref.product_id,ref.image_id,ref.src,ref.alt,ref.product_title,ref.vendor))
conn.commit()

Kontrollen

• Zähle Bilder insgesamt, protokolliere pro Batch.
• Filtere leere src und doppelte image_id.
• Prüfe, ob ONLY_EMPTY aktiv ist, wenn nur fehlende ALT Texte befüllt werden sollen.

---

2. AI ansprechen

Ziel
Aus Produktkontext prägnante, saubere ALT Texte erzeugen.

Wie

• build_context holt Material und Farben aus Varianten, plus Vendor und Tags.
• prompt_alt baut einen knappen, regelbasierten Prompt.
• call_openai_alt spricht das Modell an, hält Rate Limits ein, validiert die Antwort und kürzt auf ALT_MAX.

Qualitätsregeln

• Maximal 125 Zeichen, kein Fülltext, klare Benennung von Produktart, Material, Farbe.
• Sprache über ALT_LANG steuerbar, zum Beispiel de oder en.
• JSON Zwang in der Antwort, robustes Fallback mit Regex, falls das Modell einmal plaudert.

Fehlerbilder

• 429 oder Netzwerkfehler, der Retry ist aktiv.
• Leere Antwort, dann Abbruch mit sauberer Fehlermeldung.
• Unsupported Image Formats sind bereits vor der AI-Stufe gefiltert.

---

3. ALT-Texte speichern

Ziel
Nachvollziehbarkeit, Wiederaufnahme, Reporting.

Wie

• Jede Entscheidung landet in jobs über db_upsert.
• Spalten: alt_old, alt_new, status, error, ts.
• Typische Statuswerte: dry_run, updated, skipped_has_alt, skipped_unsupported, error.

Nützliche Abfragen

-- Alle Fehlerfälle
SELECT * FROM jobs WHERE status='error' ORDER BY ts DESC;

-- Vorschläge aus dem Dry Run prüfen
SELECT product_id,image_id,src,alt_old,alt_new
FROM jobs WHERE status='dry_run' LIMIT 100;

-- Noch offene Fälle
SELECT COUNT(*) FROM jobs
WHERE status NOT IN ('updated','skipped_has_alt','skipped_unsupported');

---

4. Updates in Shopify schreiben

Ziel
Die generierten ALT Texte zuverlässig zurück ins System bringen.

Wie

• update_image_alt sendet ein PUT auf
  /admin/api/{API_VERSION}/products/{product_id}/images/{image_id}.json
  Payload {"image":{"id":image_id,"alt":alt_new}}.
• Erfolgreiche Updates werden mit status='updated' geloggt.

Sichere Vorgehensweise

• Erst mit DRY_RUN=true fahren, Einträge stichprobenartig prüfen.
• Danach DRY_RUN=false, kleine Batches, zum Beispiel BATCH_SIZE=50.
• Rate Limits beachten, lieber konstant als parallel, dafür stabil.

Typische Fehler und Ursachen

• 404, Bild oder Produkt gelöscht.
• 403, fehlende Rechte, prüfe Admin-Token.
• 422, ungültige Payload, meist falsche ID oder Formatierung.

---

5. Datenbank synchronisieren

Ziel
DB und Shopify in Einklang halten, manuelle Änderungen erkennen, Wiederaufnahmen ohne Doppelarbeit.

Sync-Rezept

1. Pull Aktualstand
   • Vor jedem Durchlauf collect_images erneut ausführen, die aktuellen ALT Texte aus Shopify einlesen.
   • Optional die Tabelle inventory aktualisieren, siehe oben.
2. Delta ermitteln
   • Verbinde jobs und den frischen Inventarstand, erkenne manuelle Änderungen.
   • Beispiel: Ein Mitarbeiter hat im Backend einen ALT Text angepasst, dieser soll nicht überschrieben werden.

-- Manuelle Änderungen seit dem letzten Run finden
WITH last AS (
  SELECT product_id,image_id,MAX(ts) AS last_ts
  FROM jobs
  GROUP BY product_id,image_id
)
SELECT i.product_id,i.image_id,i.alt_shop AS alt_now,j.alt_new AS alt_suggest
FROM inventory i
LEFT JOIN jobs j
  ON j.product_id=i.product_id AND j.image_id=i.image_id
LEFT JOIN last l
  ON l.product_id=i.product_id AND l.image_id=i.image_id AND j.ts=l.last_ts
WHERE j.status='updated'
  AND i.alt_shop IS NOT j.alt_new;

3. Konfliktstrategie
   • Wenn alt_now ungleich alt_suggest, dann
     • entweder respektiere die manuelle Änderung und markiere status='diverged_manual',
     • oder schreibe nur zurück, wenn alt_now leer ist.
   • Regeln als Konfiguration, etwa RESPECT_MANUAL_CHANGES=true.
4. Reprocess Queue bauen
   • Offene oder fehlerhafte Jobs erneut einplanen.

SELECT product_id,image_id FROM jobs
WHERE status IN ('error')  -- oder weitere Status
ORDER BY ts ASC;

5. Idempotenz sichern
   • Das Skript überspringt bereits erledigte Einträge, siehe done Set in main.
   • Bei Wiederaufnahme bleibt die Pipeline stabil, keine Doppelupdates.

Praktische Views

CREATE VIEW v_todo AS
SELECT i.product_id,i.image_id,i.src,i.alt_shop
FROM inventory i
LEFT JOIN jobs j
  ON j.product_id=i.product_id AND j.image_id=i.image_id
WHERE j.product_id IS NULL
   OR j.status IN ('error','dry_run');

CREATE VIEW v_changed AS
SELECT i.product_id,i.image_id,i.alt_shop AS alt_now,j.alt_new
FROM inventory i
JOIN jobs j
  ON j.product_id=i.product_id AND j.image_id=i.image_id
WHERE j.status='updated' AND i.alt_shop != j.alt_new;

---

Tipps für Stabilität und Qualität

• Safety bei Zeichenlänge
  clamp_len hält 125 Zeichen ein, passe ALT_MAX nur an, wenn Deine Accessibility-Guidelines es erlauben.
• Sprache je Markt
  Nutze je Shop-Locale unterschiedliche .env Dateien, zum Beispiel ALT_LANG=en für den UK Shop.
• Kostenkontrolle
  Logge Tokens je Request, optional in einer zweiten Tabelle costs, um AI Kosten pro Batch zu sehen.

CREATE TABLE IF NOT EXISTS costs(
  ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  product_id INTEGER,
  image_id   INTEGER,
  prompt_tokens INTEGER,
  completion_tokens INTEGER
);

• Bildinhalte optional analysieren
  Wenn Du echte Bildbeschreibung statt Metadaten wünschst, lade das Bild lokal, gib der AI einen kleinen Caption-Vorschlag und kombiniere mit Produktfakten. Das erhöht Qualität, ist aber teurer. Metadaten-only ist meist ausreichend und risikoarm.
• Monitoring
  Erzeuge ein kurzes Reporting, zum Beispiel
  Anzahl verarbeitet, Anzahl Fehler, Top Fehlergründe, durchschnittliche Tokenkosten.

---

Mini Checkliste für den Live-Gang

• .env korrekt, Token getestet
• Einmal mit DRY_RUN=true, Ergebnisse prüfen
• ONLY_EMPTY=true, um bestehende gute ALT Texte nicht zu überschreiben
• Kleine Batches, Logs beobachten
• Nach dem Lauf Stichprobe im Frontend, dann in der Google Search Console Reiter Bilder beobachten

---

Vergleich: Einfaches Setup vs. Advanced Setup für ALT-Texte in Shopify

Kriterium	Einfaches Setup	Advanced Setup
Zielgruppe	Kleine Shops mit < 200 Bildern	Große Shops mit mehreren tausend Bildern
Vorgehen	ALT-Texte manuell im Shopify-Backend eintragen oder per CSV-Import anpassen	Vollautomatische Generierung & Eintragung über AI + Shopify API
Technik	Kein Code notwendig	Python-Skript, API-Zugriff, Datenbank für Jobs & Logging
AI-Einsatz	Texte einzeln in ChatGPT generieren und kopieren	AI generiert Tausende ALT-Texte in Batches mit Wiederaufnahme
Zeitaufwand	Hoch – jedes Bild wird manuell gepflegt	Sehr gering – auch 10.000 Bilder können in wenigen Stunden verarbeitet werden
Flexibilität	Kaum: Änderungen müssen händisch durchgeführt werden	Hoch: Regeln, Prompts, Sprachen, Zeichenlänge frei definierbar
Risiken	Fehlerquote durch Copy-Paste, leicht inkonsistente Texte	Technisches Setup erfordert etwas Know-how, aber dafür konsistent und skalierbar
Kosten	Keine App-Kosten, nur Zeitaufwand	API-Kosten (OpenAI) + einmaliger Setup-Aufwand

---

Das Python-Skript

Beispiel Python Scripts, das vollständige Erklärung findet ihr bald in meinem Mitglieder Bereich.

# ai_alt_texts_shopify.py
# Zweck: ALT-Texte mit AI generieren und direkt in Shopify schreiben
# Features: .env Support, Dry-Run, Wiederaufnahme, Längenlimit, sauberes Error-Handling, Batch-Verarbeitung

import os, time, json, re, sys, csv, sqlite3, math
from typing import Dict, Any, List, Optional, Tuple
from dataclasses import dataclass
from urllib.parse import urlparse

import requests
from dotenv import load_dotenv

# ========= ENV =========
load_dotenv()
SHOPIFY_SHOP   = os.getenv("SHOPIFY_SHOP")                  # shop.myshopify.com
SHOP_TOKEN     = os.getenv("SHOPIFY_TOKEN")                 # Admin API Access Token
API_VERSION    = os.getenv("API_VERSION", "2025-07")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_MODEL   = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
LANG           = os.getenv("ALT_LANG", "de")
ALT_MAX        = int(os.getenv("ALT_MAX", "125"))
BATCH_SIZE     = int(os.getenv("BATCH_SIZE", "50"))
DRY_RUN        = os.getenv("DRY_RUN", "true").lower() == "true"
ONLY_EMPTY     = os.getenv("ONLY_EMPTY", "true").lower() == "true"   # nur Bilder ohne ALT updaten
INCLUDE_VENDOR = os.getenv("INCLUDE_VENDOR", "true").lower() == "true"
RESUME_DB      = os.getenv("DB_PATH", "alt_text_jobs.db")

# Backoff für API Calls
OAI_MIN_INTERVAL = float(os.getenv("OAI_MIN_INTERVAL", "1.2"))  # Sekunden zwischen OpenAI Calls
SHOP_MAX_RETRY   = int(os.getenv("SHOP_MAX_RETRY", "3"))
OAI_MAX_RETRY    = int(os.getenv("OAI_MAX_RETRY", "3"))

# ========= OpenAI minimal Client =========
OAI_URL = "https://api.openai.com/v1/chat/completions"
OAI_HEADERS = {
    "Authorization": f"Bearer {OPENAI_API_KEY}",
    "Content-Type": "application/json",
}

# ========= Shopify REST =========
SHOP_BASE = f"https://{SHOPIFY_SHOP}/admin/api/{API_VERSION}"
SHOP_HEADERS = {
    "X-Shopify-Access-Token": SHOP_TOKEN,
    "Content-Type": "application/json",
}

@dataclass
class ImageRef:
    product_id: int
    image_id: int
    src: str
    alt: str
    product_title: str
    vendor: Optional[str]
    tags: Optional[str]
    variants: List[Dict[str, Any]]

# ========= Persistenz =========
def db_init(path: str):
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS jobs(
            product_id INTEGER,
            image_id   INTEGER,
            src        TEXT,
            alt_old    TEXT,
            alt_new    TEXT,
            status     TEXT,
            error      TEXT,
            ts         TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY(product_id, image_id)
        );
    """)
    return conn

def db_upsert(conn, ref: ImageRef, alt_new: Optional[str], status: str, error: Optional[str]):
    conn.execute("""
        INSERT INTO jobs(product_id,image_id,src,alt_old,alt_new,status,error)
        VALUES(?,?,?,?,?,?,?)
        ON CONFLICT(product_id,image_id) DO UPDATE SET
          alt_old=excluded.alt_old,
          alt_new=excluded.alt_new,
          status=excluded.status,
          error=excluded.error,
          ts=CURRENT_TIMESTAMP
    """, (ref.product_id, ref.image_id, ref.src, ref.alt, alt_new or "", status, error or ""))
    conn.commit()

# ========= Utils =========
def clean_text(s: str) -> str:
    s = re.sub(r"\s+", " ", s or "").strip()
    return s

def clamp_len(s: str, n: int) -> str:
    s = s.strip()
    return s if len(s) <= n else s[:n].rstrip()

def is_supported_image(src: str) -> bool:
    # optionaler Filter, falls du inkompatible Formate überspringen willst
    ext = urlparse(src).path.lower().rsplit(".", 1)[-1] if "." in urlparse(src).path else ""
    return ext in {"png", "jpg", "jpeg", "webp", "gif"}

def log(*a):
    print(*a, file=sys.stdout, flush=True)

# ========= Shopify: Produkte und Bilder holen =========
def shopify_get(url: str, params: Dict[str, Any] = None, retry=SHOP_MAX_RETRY) -> Dict[str, Any]:
    for i in range(retry):
        r = requests.get(url, headers=SHOP_HEADERS, params=params, timeout=60)
        if 200 <= r.status_code < 300:
            return r.json()
        time.sleep(1.5 * (i + 1))
    raise RuntimeError(f"Shopify GET failed {r.status_code} {r.text[:300]}")

def shopify_put(url: str, payload: Dict[str, Any], retry=SHOP_MAX_RETRY) -> Dict[str, Any]:
    for i in range(retry):
        r = requests.put(url, headers=SHOP_HEADERS, json=payload, timeout=60)
        if 200 <= r.status_code < 300:
            return r.json()
        time.sleep(1.5 * (i + 1))
    raise RuntimeError(f"Shopify PUT failed {r.status_code} {r.text[:300]}")

def iter_products(limit=250):
    page = 1
    while True:
        url = f"{SHOP_BASE}/products.json"
        data = shopify_get(url, {"limit": limit, "page": page})
        products = data.get("products", [])
        if not products:
            break
        for p in products:
            yield p
        page += 1

def collect_images() -> List[ImageRef]:
    refs: List[ImageRef] = []
    for p in iter_products():
        pid = p["id"]
        title = p.get("title") or ""
        vendor = p.get("vendor")
        tags = p.get("tags")
        variants = p.get("variants", [])
        for img in p.get("images", []):
            ref = ImageRef(
                product_id=pid,
                image_id=img["id"],
                src=img.get("src") or "",
                alt=img.get("alt") or "",
                product_title=title,
                vendor=vendor,
                tags=tags,
                variants=variants
            )
            refs.append(ref)
    return refs

# ========= Prompting =========
def build_context(ref: ImageRef) -> Dict[str, Any]:
    colors = []
    materials = []
    for v in ref.variants:
        for i in range(1, 4):
            o = v.get(f"option{i}")
            if not o:
                continue
            o_norm = o.lower()
            if any(k in o_norm for k in [""]):
                colors.append(o.strip())
            if any(k in o_norm for k in [""]):
                materials.append(o.strip())
    ctx = {
        "product_title": clean_text(ref.product_title),
        "vendor": clean_text(ref.vendor or ""),
        "colors": sorted(set([c for c in colors if c])),
        "materials": sorted(set([m for m in materials if m])),
        "tags": clean_text(ref.tags or "")
    }
    return ctx

def prompt_alt(ref: ImageRef) -> str:
    ctx = build_context(ref)
    vendor_part = f"{ctx['vendor']} " if INCLUDE_VENDOR and ctx["vendor"] else ""
    instr = f"""
Du schreibst prägnante ALT Texte auf {LANG}.
Regeln:
1. Maximal {ALT_MAX} Zeichen
2. Beschreibend und spezifisch
3. Keine Füllphrasen
4. Wenn sinnvoll Material und Farbe erwähnen
5. Keine Hashtags
6. Keine Wiederholung des Wortes Bild

Gib ausschließlich JSON mit dem Feld alt zurück.

Produktname: {ctx['product_title']}
Vendor: {ctx['vendor']}
Farben: {', '.join(ctx['colors']) if ctx['colors'] else 'n a'}
Material: {', '.join(ctx['materials']) if ctx['materials'] else 'n a'}
Tags: {ctx['tags'] if ctx['tags'] else 'n a'}

Beispielausgabe:
{{"alt":"{vendor_part}{ctx['product_title']} aus Sterling Silber"}}
""".strip()
    return instr

def call_openai_alt(ref: ImageRef) -> str:
    payload = {
        "model": OPENAI_MODEL,
        "messages": [
            {"role": "system", "content": "Du bist eine präzise Assistenz für barrierefreie und SEO taugliche ALT Texte."},
            {"role": "user", "content": prompt_alt(ref)},
        ],
        "temperature": 0.2,
    }
    last_err = None
    for attempt in range(1, OAI_MAX_RETRY + 1):
        try:
            t0 = time.time()
            r = requests.post(OAI_URL, headers=OAI_HEADERS, json=payload, timeout=120)
            if 200 <= r.status_code < 300:
                raw = r.json()["choices"][0]["message"]["content"]
                # JSON strikt parsen mit Fallback
                try:
                    data = json.loads(raw)
                    alt = data.get("alt", "").strip()
                except json.JSONDecodeError:
                    m = re.search(r'\"alt\"\s*:\s*\"(.*?)\"', raw, re.S)
                    alt = m.group(1).strip() if m else raw.strip()

                alt = clean_text(alt)
                alt = clamp_len(alt, ALT_MAX)
                if not alt:
                    raise ValueError("leere AI Antwort")
                dt = time.time() - t0
                # Rate Limit freundlich
                if dt < OAI_MIN_INTERVAL:
                    time.sleep(OAI_MIN_INTERVAL - dt)
                return alt
            else:
                last_err = f"OpenAI {r.status_code} {r.text[:200]}"
        except Exception as e:
            last_err = f"{type(e).__name__}: {e}"
        time.sleep(1.2 * attempt)
    raise RuntimeError(last_err or "OpenAI fehlgeschlagen")

# ========= Update in Shopify =========
def update_image_alt(ref: ImageRef, alt_new: str) -> Dict[str, Any]:
    url = f"{SHOP_BASE}/products/{ref.product_id}/images/{ref.image_id}.json"
    payload = {"image": {"id": ref.image_id, "alt": alt_new}}
    return shopify_put(url, payload)

# ========= Pipeline =========
def process_batch(conn, batch: List[ImageRef]):
    for ref in batch:
        try:
            if ONLY_EMPTY and ref.alt and ref.alt.strip():
                db_upsert(conn, ref, None, "skipped_has_alt", None)
                log(f"⏭️  Skip Produkt {ref.product_id} Bild {ref.image_id} hat bereits ALT")
                continue

            if not is_supported_image(ref.src):
                db_upsert(conn, ref, None, "skipped_unsupported", "unsupported format")
                log(f"⏭️  Skip unsupported Format {ref.src}")
                continue

            alt_new = call_openai_alt(ref)

            if DRY_RUN:
                db_upsert(conn, ref, alt_new, "dry_run", None)
                log(f"📝 DryRun ALT → {alt_new}")
                continue

            _ = update_image_alt(ref, alt_new)
            db_upsert(conn, ref, alt_new, "updated", None)
            log(f"✅ Updated Produkt {ref.product_id} Bild {ref.image_id}")
        except Exception as e:
            db_upsert(conn, ref, None, "error", str(e))
            log(f"❌ Fehler bei Produkt {ref.product_id} Bild {ref.image_id}: {e}")

def main():
    if not SHOPIFY_SHOP or not SHOP_TOKEN or not OPENAI_API_KEY:
        log("Bitte .env mit SHOPIFY_SHOP SHOPIFY_TOKEN OPENAI_API_KEY setzen")
        sys.exit(1)

    conn = db_init(RESUME_DB)
    all_refs = collect_images()
    log(f"Gefundene Bilder gesamt: {len(all_refs)}")

    # Resume Support: bereits bearbeitete überspringen
    done = set()
    for r in conn.execute("SELECT product_id, image_id FROM jobs WHERE status IN ('updated','dry_run','skipped_has_alt','skipped_unsupported')"):
        done.add((r[0], r[1]))

    todo = [r for r in all_refs if (r.product_id, r.image_id) not in done]
    log(f"Zu verarbeiten: {len(todo)}  DryRun={DRY_RUN}  OnlyEmpty={ONLY_EMPTY}")

    # Batches
    for i in range(0, len(todo), BATCH_SIZE):
        batch = todo[i:i+BATCH_SIZE]
        log(f"— Batch {i//BATCH_SIZE+1} von {math.ceil(len(todo)/BATCH_SIZE)} mit {len(batch)} Bildern")
        process_batch(conn, batch)

    log("Fertig")

if __name__ == "__main__":
    main()

---