""" BAN YARO — Rassen-Anreicherung (Wikipedia-grounded) Strategie: 1. Wikipedia-Einleitungstext abrufen (de → en Fallback) 2. Claude Haiku extrahiert Fakten NUR aus dem Quelltext 3. Kein Wikipedia-Artikel → ki_enriched=2, ki_source='none' (nicht veröffentlichen) ki_enriched-Werte: 0 = noch nicht verarbeitet 1 = angereichert (mit Wikipedia-Quelle) 2 = kein Wikipedia-Artikel gefunden, übersprungen """ import asyncio import json import logging import re import sys import os import httpx sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from database import db logger = logging.getLogger(__name__) _HAIKU_MODEL = "claude-haiku-4-5-20251001" _WP_HEADERS = {"User-Agent": "Banyaro/1.0 (https://banyaro.de; mail@banyaro.de) httpx"} # Übersetzungstabelle für englische TheDogAPI-Temperamentwörter _TEMPER_DE: dict[str, str] = { "adaptable": "anpassungsfähig", "active": "aktiv", "affectionate": "liebevoll", "agile": "agil", "alert": "wachsam", "aloof": "distanziert", "athletic": "sportlich", "bold": "kühn", "brave": "mutig", "calm": "ruhig", "careful": "behutsam", "cheerful": "fröhlich", "clever": "klug", "confident": "selbstbewusst", "courageous": "mutig", "curious": "neugierig", "devoted": "treu", "dignified": "würdevoll", "docile": "gelehrig", "dominant": "dominant", "eager": "eifrig", "eager to please": "folgsam", "energetic": "energisch", "even tempered": "ausgeglichen", "even-tempered": "ausgeglichen", "faithful": "treu", "fearless": "furchtlos", "feisty": "temperamentvoll", "friendly": "freundlich", "gentle": "sanft", "good-natured": "gutmütig", "happy": "fröhlich", "hardy": "robust", "independent": "selbstständig", "industrious": "fleißig", "intelligent": "intelligent", "intuitive": "intuitiv", "joyful": "fröhlich", "keen": "eifrig", "lively": "lebhaft", "loyal": "loyal", "obedient": "gehorsam", "outgoing": "offen", "patient": "geduldig", "playful": "verspielt", "protective": "beschützend", "quiet": "ruhig", "reserved": "zurückhaltend", "responsive": "aufmerksam", "sensitive": "sensibel", "smart": "klug", "sociable": "gesellig", "spirited": "temperamentvoll", "stubborn": "eigensinnig", "sweet": "sanft", "tenacious": "hartnäckig", "territorial": "territorial", "trainable": "lernfähig", "versatile": "vielseitig", "vigilant": "wachsam", "willful": "eigenwillig", "witty": "gewitzt", "work-focused": "arbeitsorientiert", } _TEMPER_GARBAGE = { "hunderasse", "dog breed", "breed of dog", "extinct dog breed", "dog", "hund", "rasse", } _DIRECT_FIELDS = { "beschreibung", "vorkommen_de", "groesse", "gewicht_min_kg", "gewicht_max_kg", "lebensdauer", "aktivitaet", "erfahrung", "kinder_geeignet", "wohnung_geeignet", "temperament", } _SYSTEM = ( "Du bist ein Datenprozessor für eine Hunderassen-Referenz-Datenbank. " "Extrahiere Informationen AUSSCHLIESSLICH aus dem gegebenen Quelltext. " "Setze null wenn eine Information nicht im Text steht. " "Erfinde keine Werte." ) _PROMPT = '''\ Extrahiere strukturierte Daten für die Hunderasse "{name}" aus diesem Wikipedia-Text. --- WIKIPEDIA ({lang}) --- {wiki_text} --- ENDE --- Antworte NUR mit einem JSON-Objekt. Fehlende Informationen = null. {{ "beschreibung": "3-5 informative Sätze über Charakter, Wesen und Verwendung aus dem Text. Schließe mit: Auf banyaro.app findest du weitere Informationen zu dieser Rasse.", "vorkommen_de": "1-2 Sätze zur Verbreitung in Deutschland/DACH, nur wenn im Text erwähnt, sonst null", "groesse": "klein|mittel|gross|sehr_gross oder null", "gewicht_min_kg": Zahl_oder_null, "gewicht_max_kg": Zahl_oder_null, "lebensdauer": "X-Y Jahre oder null", "aktivitaet": "niedrig|mittel|hoch|sehr_hoch oder null", "erfahrung": "anfaenger|fortgeschritten|experte oder null", "kinder_geeignet": true|false|null, "wohnung_geeignet": true|false|null, "temperament": "kommagetrennte Eigenschaften auf Deutsch aus dem Text, oder null" }} ''' def translate_temperament(text: str) -> str | None: if not text: return text parts = [p.strip() for p in text.split(",")] result = [] for part in parts: low = part.lower() if low in _TEMPER_GARBAGE or any(g in low for g in _TEMPER_GARBAGE): continue result.append(_TEMPER_DE.get(low, part)) return ", ".join(result) if result else None def _parse_json(raw: str) -> dict: try: return json.loads(raw) except json.JSONDecodeError: pass match = re.search(r"```(?:json)?\s*([\s\S]+?)\s*```", raw) if match: try: return json.loads(match.group(1)) except json.JSONDecodeError: pass match = re.search(r"\{[\s\S]+\}", raw) if match: try: return json.loads(match.group(0)) except json.JSONDecodeError: pass raise ValueError(f"Kein gültiges JSON in Antwort gefunden: {raw[:200]}") async def _fetch_wikipedia_text(name: str) -> tuple[str | None, str | None]: """Holt den Einleitungstext eines Wikipedia-Artikels (de → en Fallback). Returns: (text, lang) oder (None, None) wenn kein Artikel gefunden. """ for lang in ("de", "en"): try: async with httpx.AsyncClient(timeout=10, headers=_WP_HEADERS) as client: resp = await client.get( f"https://{lang}.wikipedia.org/w/api.php", params={ "action": "query", "titles": name, "prop": "extracts", "exintro": 1, "explaintext": 1, "format": "json", "redirects": 1, }, ) pages = resp.json().get("query", {}).get("pages", {}) for page in pages.values(): if page.get("pageid", -1) == -1: continue text = page.get("extract", "").strip() if len(text) > 150: return text[:3000], lang except Exception as e: logger.debug("Wikipedia-Text (%s) fehlgeschlagen für %s: %s", lang, name, e) return None, None async def _fetch_wikimedia_photo(name: str) -> str | None: """Sucht ein lizenzfreies Foto via Wikipedia pageimages API (de → en Fallback).""" for lang in ("de", "en"): try: async with httpx.AsyncClient(timeout=8, headers=_WP_HEADERS) as client: resp = await client.get( f"https://{lang}.wikipedia.org/w/api.php", params={ "action": "query", "titles": name, "prop": "pageimages", "format": "json", "pithumbsize": 800, "redirects": 1, }, ) pages = resp.json().get("query", {}).get("pages", {}) for page in pages.values(): if "thumbnail" in page: return page["thumbnail"]["source"] except Exception as e: logger.debug("Wikimedia-Foto (%s) fehlgeschlagen für %s: %s", lang, name, e) return None async def _haiku_complete(prompt: str) -> str: """Claude Haiku direkt aufrufen (immer Cloud, für maximale Genauigkeit).""" import anthropic key = os.getenv("ANTHROPIC_API_KEY", "") if not key: raise RuntimeError("ANTHROPIC_API_KEY nicht gesetzt") def _call(): client = anthropic.Anthropic(api_key=key) return client.messages.create( model=_HAIKU_MODEL, max_tokens=700, system=[{ "type": "text", "text": _SYSTEM, "cache_control": {"type": "ephemeral"}, }], messages=[{"role": "user", "content": prompt}], ) loop = asyncio.get_event_loop() resp = await loop.run_in_executor(None, _call) return resp.content[0].text.strip() async def enrich_breeds(limit: int = 10) -> int: """ Reichert bis zu `limit` Rassen an (ki_enriched = 0). Strategie: Wikipedia-Text holen → Haiku extrahiert Fakten. Kein Wikipedia-Artikel → ki_enriched=2, ki_source='none'. Returns: Anzahl erfolgreich angereicherter Rassen. """ with db() as conn: rassen = conn.execute( """SELECT id, name, slug, herkunft, foto_url FROM wiki_rassen WHERE ki_enriched = 0 ORDER BY name ASC LIMIT ?""", (limit,), ).fetchall() if not rassen: logger.info("Keine Rassen zur Anreicherung gefunden.") return 0 enriched_count = 0 for rasse in rassen: name = rasse["name"] rasse_id = rasse["id"] # 1. Wikipedia-Text holen wiki_text, wiki_lang = await _fetch_wikipedia_text(name) if not wiki_text: # Kein Artikel → markieren und überspringen with db() as conn: conn.execute( "UPDATE wiki_rassen SET ki_enriched=2, ki_source='none' WHERE id=?", (rasse_id,), ) logger.info("Kein Wikipedia-Artikel: %s → übersprungen", name) await asyncio.sleep(0.5) continue # 2. Haiku extrahiert Fakten aus dem Quelltext prompt = _PROMPT.format(name=name, lang=wiki_lang.upper(), wiki_text=wiki_text) try: raw = await _haiku_complete(prompt) except Exception as e: logger.error("Haiku-Anfrage fehlgeschlagen für %s: %s", name, e) await asyncio.sleep(3) continue try: data = _parse_json(raw) except ValueError as e: logger.warning("JSON-Parsing fehlgeschlagen für %s: %s", name, e) await asyncio.sleep(2) continue # 3. DB-Update updates = { k: v for k, v in data.items() if k in _DIRECT_FIELDS and v is not None } if "temperament" in updates: updates["temperament"] = translate_temperament(updates["temperament"]) updates["ki_enriched"] = 1 updates["ki_model"] = _HAIKU_MODEL updates["ki_source"] = f"wikipedia_{wiki_lang}" cols = ", ".join(f"{k}=?" for k in updates) values = list(updates.values()) + [rasse_id] try: with db() as conn: conn.execute(f"UPDATE wiki_rassen SET {cols} WHERE id=?", values) logger.info("Rasse angereichert: %s (%d Felder, WP-%s)", name, len(updates) - 2, wiki_lang.upper()) enriched_count += 1 except Exception as e: logger.error("DB-Update fehlgeschlagen für %s: %s", name, e) await asyncio.sleep(2) continue # 4. Foto holen wenn noch keins vorhanden if not rasse["foto_url"]: foto_url = await _fetch_wikimedia_photo(name) if foto_url: try: with db() as conn: conn.execute( "UPDATE wiki_rassen SET foto_url=? WHERE id=?", (foto_url, rasse_id), ) logger.info("Wikimedia-Foto gesetzt: %s", name) except Exception as e: logger.error("Foto-Update fehlgeschlagen für %s: %s", name, e) await asyncio.sleep(1) return enriched_count def reset_gemma_entries() -> int: """Setzt alle Gemma-angereicherten Einträge zurück auf ki_enriched=0.""" with db() as conn: cur = conn.execute( "UPDATE wiki_rassen SET ki_enriched=0, ki_model=NULL, ki_source=NULL " "WHERE ki_model LIKE 'gemma%'", ) count = cur.rowcount logger.info("Gemma-Reset: %d Rassen zurückgesetzt", count) return count def translate_existing_temperaments() -> int: """Übersetzt alle englischen Temperament-Felder in der DB ins Deutsche.""" _english_words = set(_TEMPER_DE.keys()) updated = 0 with db() as conn: rows = conn.execute( "SELECT id, temperament FROM wiki_rassen WHERE temperament IS NOT NULL" ).fetchall() for row in rows: original = row["temperament"] parts_lower = [p.strip().lower() for p in original.split(",")] has_english = any(p in _english_words for p in parts_lower) has_garbage = any( any(g in p for g in _TEMPER_GARBAGE) for p in parts_lower ) if not has_english and not has_garbage: continue translated = translate_temperament(original) if translated != original: conn.execute( "UPDATE wiki_rassen SET temperament=? WHERE id=?", (translated, row["id"]), ) updated += 1 logger.info("Temperament-Migration: %d Rassen übersetzt", updated) return updated if __name__ == "__main__": import argparse logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") parser = argparse.ArgumentParser(description="Rassen-Anreicherung (Wikipedia-grounded)") parser.add_argument("--limit", type=int, default=10) parser.add_argument("--reset-gemma", action="store_true", help="Gemma-Einträge zurücksetzen bevor angereichert wird") args = parser.parse_args() if args.reset_gemma: n = reset_gemma_entries() print(f"Reset: {n} Gemma-Einträge zurückgesetzt") count = asyncio.run(enrich_breeds(args.limit)) print(f"Angereichert: {count} Rassen")