418 lines
14 KiB
Python
418 lines
14 KiB
Python
"""
|
|
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")
|